diff --git a/.gitea/workflows/deploy-production.yml b/.gitea/workflows/deploy-production.yml index 3affc62a..65242b09 100644 --- a/.gitea/workflows/deploy-production.yml +++ b/.gitea/workflows/deploy-production.yml @@ -53,6 +53,7 @@ jobs: DEPLOY_ACTOR: ${{ gitea.actor }} GITEA_SERVER_URL: ${{ gitea.server_url }} GITEA_RUN_ID: ${{ gitea.run_id }} + GITEA_RUN_NUMBER: ${{ gitea.run_number }} run: | # Reporting must never change the deployment result. set +e @@ -87,7 +88,9 @@ jobs: ;; esac - run_url="${GITEA_SERVER_URL}/${DEPLOY_REPOSITORY}/actions/runs/${GITEA_RUN_ID}" + # Gitea 1.23 uses the repository run index (run_number) in web URLs, + # not the internal database ID (run_id). Keep run_id for status files. + run_url="${GITEA_SERVER_URL%/}/${DEPLOY_REPOSITORY}/actions/runs/${GITEA_RUN_NUMBER}" message="$(printf '%s\n\nРепозиторий: %s\nВетка: %s\nРевизия: %s\nИнициатор: %s\nТекущее состояние: %s\nЗапуск: %s' \ "${heading}" \ "${DEPLOY_REPOSITORY}" \ diff --git a/ops/deploy/tests/test_telegram_report.py b/ops/deploy/tests/test_telegram_report.py new file mode 100644 index 00000000..2a673766 --- /dev/null +++ b/ops/deploy/tests/test_telegram_report.py @@ -0,0 +1,121 @@ +"""Offline regression tests: python -m unittest discover -s ops/deploy/tests -v. + +Runs the actual notification shell block with mocked transports, never deployment +or Telegram requests. Requires POSIX sh (Git for Windows also works). +""" + +import json +import os +from pathlib import Path +import shutil +import subprocess +import tempfile +import textwrap +import unittest + + +ROOT = Path(__file__).resolve().parents[3] +WORKFLOW = ROOT / ".gitea/workflows/deploy-production.yml" +MOCKS = r''' +curl() { + printf 'curl\n' >> "$CAPTURE.calls" + [ "$TRANSPORT" = curl ] || return 1 + for arg in "$@"; do + case "$arg" in text=*) printf '%s' "${arg#text=}" > "$CAPTURE" ;; esac + done + printf 200 +} +wget() { + printf 'wget\n' >> "$CAPTURE.calls" + [ "$TRANSPORT" = wget ] || return 1 + for arg in "$@"; do + case "$arg" in --post-data=*) printf '%s' "${arg#--post-data=}" > "$CAPTURE" ;; esac + done +} +python3() { + printf 'python3\n' >> "$CAPTURE.calls" + cat >/dev/null + [ "$TRANSPORT" = python3 ] || return 1 + printf '%s' "$TELEGRAM_MESSAGE" > "$CAPTURE" +} +''' + + +class TelegramReportTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.workflow = WORKFLOW.read_text(encoding="utf-8") + # Extract the literal last step without requiring a YAML dependency. + cls.report = cls.workflow.split( + " - name: Send deployment report to Telegram\n", 1 + )[1] + cls.script = textwrap.dedent(cls.report.split(" run: |\n", 1)[1]) + git_sh = Path("C:/Program Files/Git/bin/sh.exe") + cls.shell = str(git_sh) if os.name == "nt" and git_sh.exists() else shutil.which("sh") + if not cls.shell: + raise RuntimeError("POSIX sh is required for the offline report tests") + + def test_context_binding_and_non_fatal_reporting(self): + self.assertIn("GITEA_RUN_NUMBER: ${{ gitea.run_number }}", self.report) + self.assertIn("GITEA_RUN_ID: ${{ gitea.run_id }}", self.report) + self.assertIn("if: always()", self.report) + self.assertIn("continue-on-error: true", self.report) + + def test_shell_syntax(self): + result = subprocess.run( + [self.shell, "-n"], input=self.script, encoding="utf-8", capture_output=True + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def run_report(self, transport, number="10", server="https://git.robotail.pro/", token="dummy"): + with tempfile.TemporaryDirectory(prefix="telegram-report-test-") as temp: + directory = Path(temp) + # Deliberately different ID and index; outcome must still use the ID. + (directory / "eusrr-production-deploy-78.status").write_text("success\n") + capture = directory / "message" + env = dict(os.environ, TRANSPORT=transport, CAPTURE=capture.as_posix(), + RUNNER_TEMP=directory.as_posix(), TELEGRAM_BOT_TOKEN=token, + TELEGRAM_CHAT_ID="dummy-chat", GITEA_RUN_ID="78", + GITEA_RUN_NUMBER=number, GITEA_SERVER_URL=server, + DEPLOY_REPOSITORY="Robotail/Corp", DEPLOY_REF="master", + DEPLOY_SHA="096807f35670241c4f2ba422e2f1c189c1ec1742", + DEPLOY_ACTOR="test") + result = subprocess.run( + [self.shell], input=MOCKS + self.script, env=env, + encoding="utf-8", capture_output=True, timeout=15 + ) + self.assertEqual(result.returncode, 0, result.stderr) + calls = Path(str(capture) + ".calls") + return (capture.read_text(encoding="utf-8") if capture.exists() else "", + calls.read_text().splitlines() if calls.exists() else [], result.stdout) + + def test_url_in_all_transports_and_fallback_order(self): + for transport, expected_calls in ( + ("curl", ["curl"]), ("wget", ["curl", "wget"]), + ("python3", ["curl", "wget", "python3"]), + ): + for number, server in (("10", "https://git.robotail.pro/"), + ("11", "https://git.robotail.pro")): + with self.subTest(transport=transport, number=number): + payload, calls, output = self.run_report(transport, number, server) + message = json.loads(payload)["text"] if transport == "wget" else payload + self.assertIn(f"https://git.robotail.pro/Robotail/Corp/actions/runs/{number}", message) + self.assertNotIn("/actions/runs/78", message) + self.assertIn("успешно завершён", message) + self.assertEqual(calls, expected_calls) + self.assertIn("Telegram deployment report sent.", output) + + def test_transport_failure_is_non_fatal(self): + payload, calls, output = self.run_report("none") + self.assertEqual(payload, "") + self.assertEqual(calls, ["curl", "wget", "python3"]) + self.assertIn("no available transport could deliver it", output) + + def test_missing_token_skips_all_transports(self): + payload, calls, output = self.run_report("curl", token="") + self.assertEqual((payload, calls), ("", [])) + self.assertIn("not configured", output) + + +if __name__ == "__main__": + unittest.main()