alexei.dolgolyov
66f152ef2c
Four root causes blocked the CI test gate; all fixed minimally: 1. test_release_provider._allow_private_urls used setenv + importlib.reload(ssrf_mod). The reload permanently rebound _ALLOW_PRIVATE=True in the module; monkeypatch.setenv undid the env var on teardown but the module attribute stayed True for the rest of the session, masking every test_ssrf*/test_ssrf_hardening case (16 failures). Switched to monkeypatch.setattr on the module attribute directly — restored cleanly on teardown. 2. _FakeResponse in test_release_provider lacked the content_length attribute and a top-level read() method that the new size-cap guards in gitea.py consult before parsing (5 failures). 3. test_gate_quiet_hours_wins_over_event_type_flag was asserting the pre-refactor gate order. evaluate_event_gate now intentionally reports EVENT_TYPE_DISABLED before QUIET_HOURS so deferrable events with the event-type flag off get dropped immediately instead of being deferred and then silently discarded at drain time. Renamed the test and inverted the expectation. 4. resolve_version() returned 0.0.0+unknown in CI because pip-wheel-built hatchling distributions ended up with METADATA missing the Version field — importlib.metadata returned None. Added __version__ = "0.8.1" to notify_bridge_server/__init__.py as a third (always-available) candidate; resolve_version() now picks the max of (installed, package, source).
243 lines
8.5 KiB
Python
243 lines
8.5 KiB
Python
"""Tests for the release provider abstraction and Gitea probe."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from notify_bridge_core.release import build_release_provider, is_valid_repo
|
|
from notify_bridge_core.release.base import (
|
|
ReleaseErrorCode,
|
|
ReleaseProviderKind,
|
|
compare_versions,
|
|
is_newer,
|
|
normalise_version,
|
|
)
|
|
from notify_bridge_core.release.gitea import GiteaReleaseProvider
|
|
|
|
|
|
# --- pure utilities ---------------------------------------------------------
|
|
|
|
|
|
def test_normalise_version_strips_v_prefix() -> None:
|
|
assert normalise_version("v1.2.3") == "1.2.3"
|
|
assert normalise_version("V1.2.3") == "1.2.3"
|
|
assert normalise_version("1.2.3") == "1.2.3"
|
|
assert normalise_version("") == ""
|
|
# Only strip ``v`` when followed by a digit — guard against names like
|
|
# ``vendor-1`` being mangled into ``endor-1``.
|
|
assert normalise_version("vendor-1") == "vendor-1"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("a", "b", "expected"),
|
|
[
|
|
("0.7.3", "0.7.2", 1),
|
|
("0.7.2", "0.7.3", -1),
|
|
("0.7.2", "0.7.2", 0),
|
|
("v0.7.3", "0.7.2", 1),
|
|
("1.0.0", "0.9.99", 1),
|
|
# Stable beats prerelease at equal numerics (tie-break).
|
|
("0.7.2-rc1", "0.7.2", -1),
|
|
("0.7.2", "0.7.2-rc1", 1),
|
|
# Implicit prerelease form ``1.0a2`` must NOT extract ``2`` as a
|
|
# third numeric segment — equal to ``1.0`` stable, then stable wins.
|
|
("1.0a2", "1.0", -1),
|
|
("", "0.0.0", 0),
|
|
],
|
|
)
|
|
def test_compare_versions(a: str, b: str, expected: int) -> None:
|
|
assert compare_versions(a, b) == expected
|
|
|
|
|
|
def test_is_newer_is_strict() -> None:
|
|
assert is_newer("0.7.3", "0.7.2") is True
|
|
assert is_newer("0.7.2", "0.7.2") is False
|
|
# A pre-release of the next minor should still be flagged as newer when
|
|
# explicitly fetched with include_prereleases=True at the provider level.
|
|
assert is_newer("0.7.3-rc1", "0.7.2") is True
|
|
|
|
|
|
def test_is_valid_repo() -> None:
|
|
assert is_valid_repo("alexei.dolgolyov/notify-bridge") is True
|
|
assert is_valid_repo("a/b") is True
|
|
assert is_valid_repo("a_b/c.d-e") is True
|
|
assert is_valid_repo("") is False
|
|
assert is_valid_repo("no-slash") is False
|
|
# Path-traversal attempts.
|
|
assert is_valid_repo("foo/bar/../admin") is False
|
|
assert is_valid_repo("foo/bar/baz") is False
|
|
assert is_valid_repo("foo/../bar") is False
|
|
# Embedded special chars.
|
|
assert is_valid_repo("foo@bar/baz") is False
|
|
assert is_valid_repo("foo/bar?x=1") is False
|
|
|
|
|
|
# --- registry ---------------------------------------------------------------
|
|
|
|
|
|
def test_registry_returns_none_for_disabled() -> None:
|
|
assert build_release_provider("disabled", session=MagicMock(), url="x", repo="a/b") is None
|
|
|
|
|
|
def test_registry_returns_none_for_unknown_kind() -> None:
|
|
assert build_release_provider("svn", session=MagicMock(), url="x", repo="a/b") is None
|
|
|
|
|
|
def test_registry_gitea_requires_url_and_valid_repo() -> None:
|
|
sess = MagicMock()
|
|
assert build_release_provider("gitea", session=sess, url="", repo="a/b") is None
|
|
assert build_release_provider("gitea", session=sess, url="https://x", repo="") is None
|
|
# Path traversal blocked by repo validation.
|
|
assert build_release_provider("gitea", session=sess, url="https://x", repo="a/b/../c") is None
|
|
provider = build_release_provider("gitea", session=sess, url="https://x", repo="a/b")
|
|
assert isinstance(provider, GiteaReleaseProvider)
|
|
assert provider.kind is ReleaseProviderKind.GITEA
|
|
|
|
|
|
# --- Gitea provider ---------------------------------------------------------
|
|
|
|
|
|
def _gitea_payload(**overrides: Any) -> list[dict[str, Any]]:
|
|
base = {
|
|
"tag_name": "v0.7.3",
|
|
"name": "v0.7.3",
|
|
"html_url": "https://git.example.com/owner/repo/releases/tag/v0.7.3",
|
|
"body": "Notes",
|
|
"published_at": "2026-05-01T00:00:00Z",
|
|
"draft": False,
|
|
"prerelease": False,
|
|
}
|
|
base.update(overrides)
|
|
return [base]
|
|
|
|
|
|
class _FakeContent:
|
|
def __init__(self, raw: bytes) -> None:
|
|
self._raw = raw
|
|
|
|
async def read(self, n: int = -1) -> bytes:
|
|
return self._raw if n < 0 else self._raw[:n]
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status: int, payload: Any) -> None:
|
|
self.status = status
|
|
import json
|
|
|
|
self.content = _FakeContent(json.dumps(payload).encode("utf-8"))
|
|
self._payload = payload
|
|
# Mimic aiohttp.ClientResponse — readers consult content_length to
|
|
# short-circuit oversized bodies before reading.
|
|
self.content_length: int | None = None
|
|
|
|
async def read(self) -> bytes:
|
|
import json
|
|
|
|
return json.dumps(self._payload).encode("utf-8")
|
|
|
|
async def json(self) -> Any:
|
|
return self._payload
|
|
|
|
async def __aenter__(self) -> "_FakeResponse":
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
|
|
def _session_with(payload: Any, status: int = 200) -> MagicMock:
|
|
"""Return a session whose `.get()` yields a fresh response per call.
|
|
|
|
Using ``side_effect`` rather than ``return_value`` ensures multiple
|
|
awaited fetches don't share mutable response state across tests.
|
|
"""
|
|
sess = MagicMock()
|
|
sess.get = MagicMock(side_effect=lambda *a, **kw: _FakeResponse(status, payload))
|
|
return sess
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _allow_private_urls(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""SSRF guard rejects example.com → publicly resolvable, so tests pass.
|
|
|
|
But we explicitly enable the bypass to remove DNS-resolution flakiness
|
|
from CI runs. Patch the module attribute directly — env-var-plus-reload
|
|
permanently mutates the module and leaks the bypass into unrelated SSRF
|
|
test files that run later in the session.
|
|
"""
|
|
import notify_bridge_core.notifications.ssrf as ssrf_mod
|
|
|
|
monkeypatch.setattr(ssrf_mod, "_ALLOW_PRIVATE", True)
|
|
|
|
|
|
async def test_gitea_fetch_latest_happy_path() -> None:
|
|
sess = _session_with(_gitea_payload())
|
|
provider = GiteaReleaseProvider(sess, "https://git.example.com/", "owner/repo")
|
|
|
|
info = await provider.fetch_latest(include_prereleases=False)
|
|
assert info is not None
|
|
assert info.tag == "v0.7.3"
|
|
assert info.version == "0.7.3"
|
|
assert info.url == "https://git.example.com/owner/repo/releases/tag/v0.7.3"
|
|
assert info.prerelease is False
|
|
|
|
|
|
async def test_gitea_skips_prereleases_by_default() -> None:
|
|
payload = _gitea_payload(prerelease=True)
|
|
sess = _session_with(payload)
|
|
provider = GiteaReleaseProvider(sess, "https://x.example.com", "a/b")
|
|
assert await provider.fetch_latest(include_prereleases=False) is None
|
|
|
|
|
|
async def test_gitea_includes_prereleases_when_asked() -> None:
|
|
payload = _gitea_payload(prerelease=True)
|
|
sess = _session_with(payload)
|
|
provider = GiteaReleaseProvider(sess, "https://x.example.com", "a/b")
|
|
info = await provider.fetch_latest(include_prereleases=True)
|
|
assert info is not None
|
|
assert info.prerelease is True
|
|
|
|
|
|
async def test_gitea_skips_drafts() -> None:
|
|
payload = _gitea_payload(draft=True)
|
|
sess = _session_with(payload)
|
|
provider = GiteaReleaseProvider(sess, "https://x.example.com", "a/b")
|
|
assert await provider.fetch_latest(include_prereleases=True) is None
|
|
|
|
|
|
async def test_gitea_returns_none_on_http_error() -> None:
|
|
sess = _session_with([], status=500)
|
|
provider = GiteaReleaseProvider(sess, "https://x.example.com", "a/b")
|
|
assert await provider.fetch_latest() is None
|
|
|
|
|
|
async def test_gitea_test_returns_structured_status() -> None:
|
|
sess = _session_with(_gitea_payload())
|
|
provider = GiteaReleaseProvider(sess, "https://x.example.com", "a/b")
|
|
result = await provider.test()
|
|
assert result["ok"] is True
|
|
assert result["info"] is not None
|
|
assert result["error"] is None
|
|
|
|
|
|
async def test_gitea_test_reports_http_error() -> None:
|
|
sess = _session_with([], status=404)
|
|
provider = GiteaReleaseProvider(sess, "https://x.example.com", "a/b")
|
|
result = await provider.test()
|
|
assert result["ok"] is False
|
|
assert result["info"] is None
|
|
# Taxonomy code, not a raw exception string.
|
|
assert result["error"] in {code.value for code in ReleaseErrorCode}
|
|
|
|
|
|
def test_gitea_constructor_validates_repo_format() -> None:
|
|
with pytest.raises(ValueError):
|
|
GiteaReleaseProvider(MagicMock(), "https://x.example.com", "no-slash")
|
|
with pytest.raises(ValueError):
|
|
GiteaReleaseProvider(MagicMock(), "https://x.example.com", "foo/bar/../baz")
|
|
with pytest.raises(ValueError):
|
|
GiteaReleaseProvider(MagicMock(), "", "owner/repo")
|