cmbot/tests/test_update_service.py

"""Tests for services.update_service — no GUI dependency."""
import json
import base64
import shutil
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from urllib import error

sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from services.update_service import (
    UpdateInfo,
    check_for_update,
    is_newer,
    load_manifest,
    parse_version,
)


class TestVersionCompare(unittest.TestCase):
    def test_parse_basic(self):
        self.assertEqual(parse_version("1.2.3"), (1, 2, 3))

    def test_parse_pads_missing_parts(self):
        self.assertEqual(parse_version("1"), (1, 0, 0))
        self.assertEqual(parse_version("1.5"), (1, 5, 0))

    def test_parse_tolerates_suffix(self):
        self.assertEqual(parse_version("1.2.3rc1"), (1, 2, 3))
        self.assertEqual(parse_version("v"), (0, 0, 0))

    def test_is_newer(self):
        self.assertTrue(is_newer("1.1.0", "1.0.0"))
        self.assertTrue(is_newer("1.0.1", "1.0.0"))
        self.assertTrue(is_newer("2.0.0", "1.9.9"))

    def test_is_not_newer(self):
        self.assertFalse(is_newer("1.0.0", "1.0.0"))
        self.assertFalse(is_newer("1.0.0", "1.1.0"))
        self.assertFalse(is_newer("0.9.9", "1.0.0"))


class TestCheckForUpdate(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp())

    def tearDown(self):
        shutil.rmtree(str(self.tmp), ignore_errors=True)

    def _write_manifest(self, data):
        with open(str(self.tmp / "manifest.json"), "w", encoding="utf-8") as f:
            json.dump(data, f)

    def test_no_source_returns_none(self):
        self.assertIsNone(check_for_update("", "1.0.0"))

    def test_missing_manifest_returns_none(self):
        # tmp exists but has no manifest.json
        self.assertIsNone(check_for_update(str(self.tmp), "1.0.0"))

    def test_unreachable_source_returns_none(self):
        self.assertIsNone(check_for_update(str(self.tmp / "nope"), "1.0.0"))

    def test_malformed_manifest_returns_none(self):
        with open(str(self.tmp / "manifest.json"), "w", encoding="utf-8") as f:
            f.write("{ not valid json")
        self.assertIsNone(check_for_update(str(self.tmp), "1.0.0"))

    def test_non_object_manifest_returns_none(self):
        self._write_manifest(["1.1.0"])
        self.assertIsNone(check_for_update(str(self.tmp), "1.0.0"))

    def test_newer_version_returns_info(self):
        self._write_manifest({
            "version": "1.1.0",
            "source": r"\\nas\cmbot\releases\CMBot-1.1.0",
            "notes": "fix batch export",
            "mandatory": False,
        })
        info = check_for_update(str(self.tmp), "1.0.0")
        self.assertIsInstance(info, UpdateInfo)
        self.assertEqual(info.version, "1.1.0")
        self.assertEqual(info.source, r"\\nas\cmbot\releases\CMBot-1.1.0")
        self.assertEqual(info.notes, "fix batch export")
        self.assertFalse(info.mandatory)

    def test_same_version_returns_none(self):
        self._write_manifest({"version": "1.0.0"})
        self.assertIsNone(check_for_update(str(self.tmp), "1.0.0"))

    def test_older_version_returns_none(self):
        self._write_manifest({"version": "0.9.0"})
        self.assertIsNone(check_for_update(str(self.tmp), "1.0.0"))

    def test_missing_version_field_returns_none(self):
        self._write_manifest({"notes": "no version here"})
        self.assertIsNone(check_for_update(str(self.tmp), "1.0.0"))

    def test_local_manifest_with_utf8_bom(self):
        # PowerShell Set-Content -Encoding UTF8 emits a BOM; must still parse.
        with open(str(self.tmp / "manifest.json"), "w", encoding="utf-8-sig") as f:
            json.dump({"version": "1.1.0"}, f)
        info = check_for_update(str(self.tmp), "1.0.0")
        self.assertIsInstance(info, UpdateInfo)
        self.assertEqual(info.version, "1.1.0")

    def test_load_manifest_returns_dict(self):
        self._write_manifest({"version": "1.2.0", "notes": "x"})
        data = load_manifest(str(self.tmp))
        self.assertEqual(data["version"], "1.2.0")

    def test_load_manifest_raises_when_missing(self):
        # Unlike check_for_update, load_manifest surfaces the error for the test button.
        with self.assertRaises(OSError):
            load_manifest(str(self.tmp / "nope"))

    def test_source_falls_back_to_update_source(self):
        self._write_manifest({"version": "2.0.0"})   # no "source" field
        info = check_for_update(str(self.tmp), "1.0.0")
        self.assertEqual(info.source, str(self.tmp))

    def test_http_source_appends_manifest_name(self):
        calls = {}

        def fake_urlopen(req, timeout=0):
            calls["url"] = req.full_url
            return _FakeResponse({
                "version": "1.2.0",
                "url": "https://example.test/CMBot-1.2.0.zip",
                "notes": "http update",
            })

        with patch("services.update_service.request.urlopen", fake_urlopen):
            info = check_for_update("https://example.test/releases", "1.0.0")

        self.assertIsInstance(info, UpdateInfo)
        self.assertEqual(calls["url"], "https://example.test/releases/manifest.json")
        self.assertEqual(info.source, "https://example.test/CMBot-1.2.0.zip")

    def test_http_source_accepts_full_manifest_url(self):
        calls = {}

        def fake_urlopen(req, timeout=0):
            calls["url"] = req.full_url
            return _FakeResponse({"version": "1.2.0"})

        manifest_url = "https://example.test/releases/manifest.json"
        with patch("services.update_service.request.urlopen", fake_urlopen):
            info = check_for_update(manifest_url, "1.0.0")

        self.assertIsInstance(info, UpdateInfo)
        self.assertEqual(calls["url"], manifest_url)
        self.assertEqual(info.source, manifest_url)

    def test_http_basic_auth_header(self):
        calls = {}

        def fake_urlopen(req, timeout=0):
            calls["auth"] = req.get_header("Authorization")
            return _FakeResponse({"version": "2.0.0"})

        with patch("services.update_service.request.urlopen", fake_urlopen):
            check_for_update(
                "https://example.test",
                "1.0.0",
                update_user="readonly",
                update_pass="secret",
            )

        token = base64.b64encode(b"readonly:secret").decode("ascii")
        self.assertEqual(calls["auth"], "Basic {}".format(token))

    def test_http_error_returns_none(self):
        def fake_urlopen(req, timeout=0):
            raise error.URLError("offline")

        with patch("services.update_service.request.urlopen", fake_urlopen):
            self.assertIsNone(check_for_update("https://example.test", "1.0.0"))

    def test_http_malformed_json_returns_none(self):
        def fake_urlopen(req, timeout=0):
            return _FakeResponse("{ not valid json", raw=True)

        with patch("services.update_service.request.urlopen", fake_urlopen):
            self.assertIsNone(check_for_update("https://example.test", "1.0.0"))

    def test_http_manifest_with_utf8_bom(self):
        def fake_urlopen(req, timeout=0):
            return _FakeResponse('{"version": "2.0.0"}', raw=True)

        with patch("services.update_service.request.urlopen", fake_urlopen):
            info = check_for_update("https://example.test", "1.0.0")
        self.assertIsInstance(info, UpdateInfo)
        self.assertEqual(info.version, "2.0.0")


class _FakeResponse:
    def __init__(self, data, raw=False):
        if raw:
            self._payload = data.encode("utf-8")
        else:
            self._payload = json.dumps(data).encode("utf-8")

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def read(self):
        return self._payload


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