# /****************************************************************************
#  * <Novell-copyright>
#  * Copyright (c) 2001 Novell, Inc. All Rights Reserved.
#  *
#  * This program is free software; you can redistribute it and/or
#  * modify it under the terms of version 2 of the GNU General Public License
#  * as published by the Free Software Foundation.
#  *
#  * This program is distributed in the hope that it will be useful,
#  * but WITHOUT ANY WARRANTY; without even the implied warranty of
#  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  * GNU General Public License for more details.
#  *
#  * You should have received a copy of the GNU General Public License
#  * along with this program; if not, contact Novell, Inc.
#  *
#  * To contact Novell about this file by physical or electronic mail, you
#  * may find current contact information at www.novell.com.
#  * </Novell-copyright>
#  ****************************************************************************/

import io
from pathlib import Path
import tempfile
from types import SimpleNamespace
import unittest
from unittest import mock

from bongo import table
from bongo.Console import wrap
from bongo.external.simpletal import simpleTAL, simpleTALES
from bongo.store.StoreClient import CalendarACL, StoreClient
from bongo.storetool.CalendarCommands import (_collection_name,
                                               _event_filename,
                                               _event_payload)
from bongo.storetool.Connection import connect_store
from bongo.storetool.ContactCommands import AddressbookContactsCommand
from bongo.storetool.ExportMailbox import MaildirMailbox, MboxMailbox


class _Response:
    def __init__(self, code, message=""):
        self.code = code
        self.message = message


class _Stream:
    def __init__(self, responses):
        self.responses = list(responses)
        self.lines = []
        self.raw = []

    def Write(self, value):
        self.lines.append(value)

    def WriteRaw(self, value):
        self.raw.append(value)

    def GetResponse(self):
        return self.responses.pop(0)


class Python3RuntimeTests(unittest.TestCase):
    @mock.patch("bongo.storetool.Connection.StoreClient")
    def test_storetool_connection_uses_every_global_option(self, client):
        options = SimpleNamespace(
            user="alice", store="alice-store", host="store.example.test",
            port="1689", password="secret")

        connect_store(options)
        client.assert_called_once_with(
            "alice", "alice-store", host="store.example.test", port=1689,
            authPassword="secret")

    def test_calendar_import_splits_events_and_normalizes_collection_names(self):
        import vobject

        source = vobject.readOne(
            "BEGIN:VCALENDAR\r\n"
            "VERSION:2.0\r\n"
            "BEGIN:VEVENT\r\n"
            "UID:event@example.test\r\n"
            "DTSTART:20260724T080000Z\r\n"
            "END:VEVENT\r\n"
            "END:VCALENDAR\r\n")
        event = source.vevent
        payload = _event_payload(source, event)
        self.assertEqual(_collection_name("/calendars/personal"), "personal")
        self.assertIn("BEGIN:VEVENT", payload)
        self.assertIn("UID:event@example.test", payload)
        self.assertRegex(_event_filename(event), r"^import-[0-9a-f]{64}$")

    def test_contact_listing_reads_document_bodies(self):
        addressbook = SimpleNamespace(uid="addressbook-guid")
        contact = SimpleNamespace(uid="contact-guid")
        store = mock.Mock()
        store.List.side_effect = [
            [SimpleNamespace(filename="/addressbook/personal",
                             uid=addressbook.uid)],
            [contact],
        ]
        store.Read.return_value = b'{"fn": "\\u00c4nne Example"}'

        command = AddressbookContactsCommand()
        with mock.patch(
                "bongo.storetool.ContactCommands.connect_store",
                return_value=store), \
                mock.patch("builtins.print") as output:
            command.Run(SimpleNamespace(), ["personal"])

        store.List.assert_has_calls([
            mock.call("/addressbook"),
            mock.call(addressbook.uid),
        ])
        store.Read.assert_called_once_with(contact.uid)
        self.assertIn("\u00c4nne Example", output.call_args.args[0])

    def test_console_table_and_unicode_template_are_text(self):
        self.assertIsInstance(wrap("Grüße 世界", width=8), str)
        self.assertEqual(table.format_table(["Name"], [["Müller"]]),
                         "Name\n------\nMüller")

        context = simpleTALES.Context()
        context.addGlobal("name", "A&B")
        template = simpleTAL.compileHTMLTemplate(
            '<p tal:content="name">placeholder</p>')
        output = io.StringIO()
        template.expand(context, output)
        self.assertEqual(output.getvalue(), "<p>A&amp;B</p>")

    def test_store_wire_lengths_are_encoded_byte_lengths(self):
        client = object.__new__(StoreClient)
        client.stream = _Stream([
            _Response(2002), _Response(1000, "guid ok"),
            _Response(2002), _Response(1000),
            _Response(2002), _Response(1000, "guid ok"),
        ])

        client.Write("/mail/INBOX", 2, "Grüße")
        client.PropSet("guid", "display", "Grüße")
        client.Replace("guid", "Grüße")

        expected = "Grüße".encode("utf-8")
        self.assertIn(" 7", client.stream.lines[0])
        self.assertTrue(client.stream.lines[1].endswith(" 7"))
        self.assertTrue(client.stream.lines[2].endswith(" 7"))
        self.assertEqual(client.stream.raw, [expected, expected, expected])

    def test_store_info_treats_missing_document_as_normal_absence(self):
        client = object.__new__(StoreClient)
        client.stream = _Stream([_Response(4224, "Document doesn't exist")])
        self.assertIsNone(client.Info("/events/new-event"))

    def test_calendar_share_token_uses_python3_hashlib(self):
        token = CalendarACL(None)._GetUid("calendar")
        self.assertEqual(len(token), 32)
        int(token, 16)

    def test_mailbox_writers_accept_store_bytes(self):
        message = (
            b"From: sender@example.test\r\n"
            b"To: user@example.test\r\n"
            b"Subject: Python 3\r\n\r\nBody\r\n")
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            mbox_path = root / "mailbox"
            writer = MboxMailbox(str(mbox_path))
            writer.add(message)
            writer.write()
            self.assertIn(b"Subject: Python 3", mbox_path.read_bytes())

            maildir_path = root / "maildir"
            writer = MaildirMailbox(str(maildir_path))
            writer.add(message)
            writer.write()
            payloads = [
                path.read_bytes()
                for folder in ("new", "cur")
                for path in (maildir_path / folder).iterdir()
            ]
            self.assertEqual(len(payloads), 1)
            self.assertIn(b"Subject: Python 3", payloads[0])


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