VodokanalBot/tests/test_start_meshcore_ha.py

import contextlib
import io
import unittest
from unittest.mock import patch

from cyr2lat import cyr2lat
from start_meshcore_ha import (
    parse_meshcore_cyr2lat_mode,
    parse_meshcore_chunk_delay_ms,
    parse_meshcore_message_limit,
    prepare_meshcore_messages,
    send_meshcore_message,
)


class FakeResponse:
    status_code = 200


class FakeSession:
    def __init__(self):
        self.messages = []

    def get(self, url, params):
        self.messages.append(params['message'])
        return FakeResponse()


class MeshCoreMessageTest(unittest.TestCase):
    def test_cyr2lat_transliterates_russian_text(self):
        self.assertEqual(
            cyr2lat("Авария: улица Южная", mode='full'),
            "Avariya: ulitsa Yuzhnaya",
        )

    def test_cyr2lat_soft_mode_replaces_only_similar_letters(self):
        self.assertEqual(
            cyr2lat("Авария: улица Южная", mode='soft'),
            "Aвapия: yлицa Южнaя",
        )

    def test_cyr2lat_off_mode_keeps_original_text(self):
        self.assertEqual(
            cyr2lat("Авария: улица Южная", mode='off'),
            "Авария: улица Южная",
        )

    def test_short_message_is_sent_as_single_payload(self):
        session = FakeSession()

        with contextlib.redirect_stdout(io.StringIO()):
            send_meshcore_message(session, "Авария", 133)

        self.assertEqual(session.messages, ["Aвapия"])

    def test_long_message_is_split_by_byte_limit(self):
        limit = 30
        messages = prepare_meshcore_messages("улица Южная " * 10, limit)

        self.assertGreater(len(messages), 1)
        for message in messages:
            self.assertLessEqual(len(message.encode("utf-8")), limit)

    def test_prepare_messages_can_skip_cyr2lat(self):
        self.assertEqual(
            prepare_meshcore_messages("Авария", 133, cyr2lat_mode='off'),
            ["Авария"],
        )

    def test_chunk_prefix_is_counted_inside_limit(self):
        limit = 12
        messages = prepare_meshcore_messages("aa bb cc dd ee ff gg hh", limit)

        self.assertGreater(len(messages), 1)
        for index, message in enumerate(messages, start=1):
            self.assertTrue(
                message.startswith("[" + str(index) + "/" + str(len(messages)) + "] ")
            )
            self.assertLessEqual(len(message.encode("utf-8")), limit)

    def test_delay_is_applied_between_chunks_only(self):
        session = FakeSession()

        with contextlib.redirect_stdout(io.StringIO()):
            with patch('start_meshcore_ha.time.sleep') as sleep:
                send_meshcore_message(session, "aa bb cc dd ee ff gg hh", 12, chunk_delay_ms=250)

        self.assertGreater(len(session.messages), 1)
        self.assertEqual(sleep.call_count, len(session.messages) - 1)
        sleep.assert_called_with(0.25)

    def test_long_word_is_split_without_breaking_utf8_characters(self):
        limit = 15
        messages = prepare_meshcore_messages("😀" * 10, limit)

        self.assertGreater(len(messages), 1)
        for message in messages:
            self.assertLessEqual(len(message.encode("utf-8")), limit)

    def test_env_limit_defaults_to_133_bytes(self):
        with patch.dict('os.environ', {}, clear=True):
            self.assertEqual(parse_meshcore_message_limit(), 133)

    def test_env_limit_rejects_too_small_value(self):
        with patch.dict('os.environ', {'MESHCORE_MESSAGE_LIMIT_BYTES': '49'}):
            with self.assertRaises(ValueError):
                parse_meshcore_message_limit()

    def test_env_limit_rejects_non_integer_value(self):
        with patch.dict('os.environ', {'MESHCORE_MESSAGE_LIMIT_BYTES': 'small'}):
            with self.assertRaises(ValueError):
                parse_meshcore_message_limit()

    def test_env_cyr2lat_mode_defaults_to_soft(self):
        with patch.dict('os.environ', {}, clear=True):
            self.assertEqual(parse_meshcore_cyr2lat_mode(), 'soft')

    def test_env_cyr2lat_mode_accepts_full(self):
        with patch.dict('os.environ', {'MESHCORE_CYR2LAT_MODE': 'full'}):
            self.assertEqual(parse_meshcore_cyr2lat_mode(), 'full')

    def test_env_cyr2lat_mode_accepts_off(self):
        with patch.dict('os.environ', {'MESHCORE_CYR2LAT_MODE': 'off'}):
            self.assertEqual(parse_meshcore_cyr2lat_mode(), 'off')

    def test_env_cyr2lat_mode_rejects_unknown_value(self):
        with patch.dict('os.environ', {'MESHCORE_CYR2LAT_MODE': 'mixed'}):
            with self.assertRaises(ValueError):
                parse_meshcore_cyr2lat_mode()

    def test_env_chunk_delay_defaults_to_zero_ms(self):
        with patch.dict('os.environ', {}, clear=True):
            self.assertEqual(parse_meshcore_chunk_delay_ms(), 0)

    def test_env_chunk_delay_accepts_positive_integer(self):
        with patch.dict('os.environ', {'MESHCORE_CHUNK_DELAY_MS': '250'}):
            self.assertEqual(parse_meshcore_chunk_delay_ms(), 250)

    def test_env_chunk_delay_rejects_negative_value(self):
        with patch.dict('os.environ', {'MESHCORE_CHUNK_DELAY_MS': '-1'}):
            with self.assertRaises(ValueError):
                parse_meshcore_chunk_delay_ms()

    def test_env_chunk_delay_rejects_non_integer_value(self):
        with patch.dict('os.environ', {'MESHCORE_CHUNK_DELAY_MS': 'slow'}):
            with self.assertRaises(ValueError):
                parse_meshcore_chunk_delay_ms()


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