Add Drone Discord changelog pipeline and test notification

tests/test_discord_changelog_ci.py

@@ -0,0 +1,143 @@
+"""Exercise changelog selection against real push ranges and mock Discord delivery."""
+import contextlib
+import importlib.util
+import io
+import json
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+import urllib.error
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location('discord_changelog_ci', ROOT / 'tools/discord_changelog_ci.py')
+ci = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(ci)
+OLD = '# Changelog\n\n## Older update\n\n- Older fix.\n'
+NEW = '# Changelog\n\n## Latest update\n\n- New fix.\n\n### Gameplay\n\n- New feature.\n\n## Older update\n\n- Older fix.\n'
+URL = 'https://discord.com/api/webhooks/123/test-token'
+
+
+class SelectionTests(unittest.TestCase):
+    def test_only_latest_section(self):
+        message = ci.latest_changes(OLD, NEW)
+        self.assertIn('New feature.', message)
+        self.assertIn('### Gameplay', message)
+        self.assertNotIn('Older fix.', message)
+        self.assertEqual(message, ci.latest_changes('', NEW.replace('\n', '\r\n')))
+
+    def test_skip_unchanged_or_historical_edits(self):
+        self.assertIsNone(ci.latest_changes(NEW, NEW))
+        self.assertIsNone(ci.latest_changes(NEW, NEW.replace('Older fix.', 'Corrected old note.')))
+        self.assertIsNone(ci.latest_changes(NEW, OLD))
+
+    def test_edit_current_entry(self):
+        self.assertIn('Corrected feature.', ci.latest_changes(NEW, NEW.replace('New feature.', 'Corrected feature.')))
+
+    def test_invalid_entries(self):
+        for text in ('', '# Changelog', '## Empty\n', '## A\nOne\n## A\nTwo', '## Huge\n' + 'x' * 2000,
+                     '## Emoji\n' + '\U0001f680' * 1000):
+            with self.subTest(text=text[:25]), self.assertRaises(ci.ChangelogError):
+                ci.latest_changes('', text)
+
+    def test_exact_message_limit(self):
+        text = '## A\n' + 'x' * 1993
+        self.assertEqual(len(ci.latest_changes('', text)), 2000)
+
+
+class PushTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.repo = Path(self.temp.name)
+        self.git('init', '-q')
+        self.git('config', 'user.name', 'Changelog test')
+        self.git('config', 'user.email', 'test@example.invalid')
+        self.first = self.save('README.txt', 'Initial checkout')
+
+    def git(self, *args):
+        return subprocess.check_output(['git', *args], cwd=self.repo, text=True, stderr=subprocess.PIPE).strip()
+
+    def save(self, name, text):
+        (self.repo / name).write_text(text)
+        self.git('add', name)
+        self.git('commit', '-qm', 'Fixture update')
+        return self.git('rev-parse', 'HEAD')
+
+    def env(self, before, after):
+        return dict(DRONE_BUILD_EVENT='push', DRONE_COMMIT_BRANCH='main',
+                    DRONE_COMMIT_BEFORE=before, DRONE_COMMIT_SHA=after)
+
+    def test_first_changelog_and_new_branch(self):
+        after = self.save('CHANGELOG.md', NEW)
+        self.assertEqual(ci.prepare(self.repo, self.env(self.first, after)), ci.latest_changes('', NEW))
+        self.assertEqual(ci.prepare(self.repo, self.env('0' * 40, after)), ci.latest_changes('', NEW))
+
+    def test_whole_push_not_only_last_commit(self):
+        before = self.save('CHANGELOG.md', OLD)
+        self.save('CHANGELOG.md', NEW)
+        after = self.save('README.txt', 'Documentation follows changelog in same push')
+        self.assertEqual(ci.prepare(self.repo, self.env(before, after)), ci.latest_changes(OLD, NEW))
+        next_push = self.save('README.txt', 'Unrelated next push')
+        self.assertIsNone(ci.prepare(self.repo, self.env(after, next_push)))
+
+    def test_checkout_mismatch_and_missing_history(self):
+        after = self.save('CHANGELOG.md', NEW)
+        for before, tip in ((self.first, self.first), ('f' * 40, after), ('', after), ('--help', after)):
+            with self.subTest(before=before, tip=tip), self.assertRaises(ci.ChangelogError):
+                ci.prepare(self.repo, self.env(before, tip))
+
+    def test_other_events_skip_before_reading_git_or_secret(self):
+        for event, branch in (('pull_request', 'main'), ('tag', 'main'), ('push', 'feature')):
+            self.assertIsNone(ci.prepare(Path('/does-not-exist'), dict(DRONE_BUILD_EVENT=event, DRONE_COMMIT_BRANCH=branch)))
+
+
+class DeliveryTests(unittest.TestCase):
+    @patch.object(ci.urllib.request, 'build_opener')
+    def test_confirmed_payload_mentions_disabled(self, opener):
+        opener.return_value.open.return_value.__enter__.return_value = io.BytesIO(b'{"id":"456"}')
+        ci.send('- Fixed @everyone notification.', URL)
+        request = opener.return_value.open.call_args.args[0]
+        self.assertEqual(request.full_url, URL + '?wait=true')
+        self.assertEqual(json.loads(request.data), {
+            'content': '- Fixed @everyone notification.', 'allowed_mentions': {'parse': []}})
+        self.assertEqual(opener.return_value.open.call_args.kwargs['timeout'], 30)
+
+    @patch.object(ci.urllib.request, 'build_opener')
+    def test_failures_never_retry_or_leak_url(self, opener):
+        errors = [urllib.error.HTTPError(URL, 429, 'limited', {}, None),
+                  urllib.error.URLError(URL), TimeoutError(URL)]
+        for error in errors:
+            opener.return_value.open.reset_mock()
+            opener.return_value.open.side_effect = error
+            with self.assertRaises(ci.ChangelogError) as raised:
+                ci.send('- Fix.', URL)
+            self.assertNotIn('test-token', str(raised.exception))
+            self.assertEqual(opener.return_value.open.call_count, 1)
+
+    @patch.object(ci.urllib.request, 'build_opener')
+    def test_no_confirmation_is_failure(self, opener):
+        for response in (b'{}', b'[]', b'not json'):
+            opener.return_value.open.return_value.__enter__.return_value = io.BytesIO(response)
+            with self.assertRaises(ci.ChangelogError):
+                ci.send('- Fix.', URL)
+
+    def test_redirects_and_invalid_destinations(self):
+        self.assertIsNone(ci.NoRedirect().redirect_request(None, None, 302, '', {}, 'https://example.invalid'))
+        for url in ('', URL.replace('https:', 'http:'), URL.replace('discord.com', 'example.invalid'),
+                    URL + '?wait=false', URL + '#fragment'):
+            with self.subTest(url=url), self.assertRaises(ci.ChangelogError):
+                ci.send('- Fix.', url)
+
+    @patch.object(ci, 'send')
+    @patch.object(ci, 'prepare', return_value='- Preview only.')
+    def test_dry_run_does_not_send(self, prepare, send):
+        with contextlib.redirect_stdout(io.StringIO()) as output:
+            self.assertEqual(ci.main(['--dry-run']), 0)
+        self.assertEqual(output.getvalue().strip(), '- Preview only.')
+        send.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()