158 lines
7.6 KiB
Python
158 lines
7.6 KiB
Python
"""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_workshop_version_context_and_no_metadata_only_repost(self):
|
|
with_version = NEW.replace('# Changelog\n', '# Changelog\n\nLatest published Workshop version: **0.0.109**\n')
|
|
self.assertIn('Last published Workshop version: **0.0.109**', ci.latest_changes(OLD, with_version))
|
|
self.assertIsNone(ci.latest_changes(NEW, with_version))
|
|
self.assertIsNone(ci.latest_changes(with_version, with_version.replace('0.0.109', '0.0.110')))
|
|
for bad in ('not-a-version', '99999.0.0'):
|
|
with self.assertRaises(ci.ChangelogError):
|
|
ci.latest_changes(OLD, with_version.replace('0.0.109', bad))
|
|
|
|
def test_version_note_counts_toward_discord_limit(self):
|
|
text = '# Changelog\n\nLatest published Workshop version: **0.0.109**\n\n## A\n' + 'x' * 1993
|
|
with self.assertRaises(ci.ChangelogError):
|
|
ci.latest_changes('', text)
|
|
|
|
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()
|