Fix Gitea run URL in Telegram deployment reports
All checks were successful
Deploy production / deploy (push) Successful in 10m54s
All checks were successful
Deploy production / deploy (push) Successful in 10m54s
This commit is contained in:
parent
096807f356
commit
0f2b181d7d
@ -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}" \
|
||||
|
||||
121
ops/deploy/tests/test_telegram_report.py
Normal file
121
ops/deploy/tests/test_telegram_report.py
Normal file
@ -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()
|
||||
Loading…
x
Reference in New Issue
Block a user