64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import runpy
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class CapturedInput(io.BytesIO):
|
|
def close(self):
|
|
self.captured = self.getvalue()
|
|
super().close()
|
|
|
|
|
|
class UploadTransportTest(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
self.folder = Path(directory.name)
|
|
self.payload = b'verified fixture JAR'
|
|
(self.folder / 'werkjournal.jar').write_bytes(self.payload)
|
|
digest = hashlib.sha256(self.payload).hexdigest()
|
|
(self.folder / 'manifest.json').write_text(json.dumps({
|
|
'schemaVersion': 1, 'runtime': 'java-25', 'commit': 'a' * 40,
|
|
'sha256': digest, 'files': {'werkjournal.jar': digest}}))
|
|
|
|
def invoke(self):
|
|
with patch.object(sys, 'argv', ['deploy-prototype.py', 'user@host', str(self.folder), '--upload']), \
|
|
patch.dict(os.environ, {}, clear=True), contextlib.redirect_stdout(io.StringIO()):
|
|
runpy.run_path(str(ROOT / 'scripts/deploy-prototype.py'), run_name='__main__')
|
|
|
|
def test_upload_sends_exact_artifact_once_and_selects_non_deploy_mode(self):
|
|
stream = CapturedInput()
|
|
process = type('Process', (), {'stdin': stream, 'wait': lambda self: 0})()
|
|
with patch.object(subprocess, 'Popen', return_value=process) as start:
|
|
with self.assertRaises(SystemExit) as exit:
|
|
self.invoke()
|
|
self.assertEqual(0, exit.exception.code)
|
|
start.assert_called_once()
|
|
command = start.call_args.args[0]
|
|
self.assertEqual(['ssh', '-o', 'BatchMode=yes', 'user@host'], command[:4])
|
|
self.assertEqual('--upload', shlex.split(command[4])[-1])
|
|
with tarfile.open(fileobj=io.BytesIO(stream.captured)) as archive:
|
|
self.assertEqual(['manifest.json', 'werkjournal.jar'], archive.getnames())
|
|
self.assertEqual(self.payload, archive.extractfile('werkjournal.jar').read())
|
|
self.assertEqual((self.folder / 'manifest.json').read_bytes(), archive.extractfile('manifest.json').read())
|
|
|
|
def test_bad_local_artifact_is_rejected_before_ssh(self):
|
|
(self.folder / 'werkjournal.jar').write_bytes(b'changed')
|
|
with patch.object(subprocess, 'Popen') as start:
|
|
with self.assertRaisesRegex(SystemExit, 'checksum'):
|
|
self.invoke()
|
|
start.assert_not_called()
|