school_compare/scripts/ci/tests/test_release.py

import json
from io import BytesIO
from urllib.error import HTTPError, URLError
from unittest.mock import Mock
import pytest
from scripts.ci import release

SHA = 'a' * 40
BUILD = 'b' * 32
DIGESTS = ['sha256:' + c * 64 for c in '123']


@pytest.fixture
def docker(monkeypatch):
    monkeypatch.setenv('REGISTRY', 'registry.example')
    refs = {}
    for component, digest in zip(release.COMPONENTS, DIGESTS):
        monkeypatch.setenv(component + '_IMAGE_NAME', component.lower())
        monkeypatch.setenv(component + '_DIGEST', digest)
        refs[f'registry.example/{component.lower()}'] = digest
    def run(*args):
        if args[0] == 'create': return ''
        if args[-1] == '{{json .Manifest}}':
            return json.dumps({'digest': refs[args[1].split(':')[0]]})
        return json.dumps({'config': {'Labels': {'io.schoolcompare.commit': SHA,
                                                'io.schoolcompare.build-id': BUILD}}})
    mock = Mock(side_effect=run)
    monkeypatch.setattr(release, 'docker', mock)
    return mock


def test_wrong_deployed_build_is_rejected_even_at_same_commit():
    assert not release.matches({'frontend': {'sha': SHA, 'build_id': BUILD},
                                'backend': {'sha': SHA, 'build_id': 'c' * 32}}, SHA, BUILD)
    assert release.matches({c: {'sha': SHA, 'build_id': BUILD} for c in ('frontend', 'backend')}, SHA, BUILD)


def test_verification_tags_the_captured_digests(docker):
    release.verify(SHA, BUILD)
    creates = [c.args for c in docker.call_args_list if c.args[0] == 'create']
    assert len(creates) == 3
    for call, digest in zip(creates, DIGESTS):
        assert call[-1].endswith('@' + digest)
        assert call[2].endswith(':verified-' + SHA)


def test_promotion_resolves_all_verified_images_before_mutation(docker):
    result = release.promote(SHA)
    assert result['build_id'] == BUILD
    calls = [c.args for c in docker.call_args_list]
    first_write = next(i for i, c in enumerate(calls) if c[0] == 'create')
    assert first_write == 6  # each of three candidates needs manifest + config
    assert all(c[-1].endswith('@' + d) for c, d in zip(calls[-3:], DIGESTS))


def test_mixed_builds_fail_before_any_tag_is_changed(docker, monkeypatch):
    identities = iter([(SHA, BUILD), (SHA, 'c' * 32), (SHA, BUILD)])
    monkeypatch.setattr(release, 'image_identity', lambda _: next(identities))
    with pytest.raises(ValueError, match='mixed'):
        release.promote(SHA)
    assert not any(c.args[0] == 'create' for c in docker.call_args_list)


def test_missing_candidate_fails_before_any_tag_is_changed(docker):
    docker.side_effect = RuntimeError('missing verified tag')
    with pytest.raises(RuntimeError): release.promote(SHA)
    assert not any(c.args[0] == 'create' for c in docker.call_args_list)


@pytest.fixture
def poll(monkeypatch):
    now = [0.0]
    monkeypatch.setattr(release.time, 'monotonic', lambda: now[0])
    monkeypatch.setattr(release.time, 'sleep', lambda seconds: now.__setitem__(0, now[0] + seconds))
    opener = Mock()
    monkeypatch.setattr(release, 'urlopen', opener)
    return opener


def response(payload):
    return BytesIO(json.dumps(payload).encode())


def test_wait_identifies_its_client_and_retries_until_both_services_match(poll, capsys):
    poll.side_effect = [
        HTTPError('https://secret.example', 503, 'unavailable', {}, None),
        response({'frontend': {'sha': SHA, 'build_id': BUILD},
                  'backend': {'sha': SHA, 'build_id': 'c' * 32}}),
        response({component: {'sha': SHA, 'build_id': BUILD}
                  for component in ('frontend', 'backend')}),
    ]
    release.wait('https://secret.example/', SHA, BUILD, 15)
    assert poll.call_count == 3
    request = poll.call_args.args[0]
    assert request.get_header('User-agent') == 'SchoolCompare-Release-Check/1.0'
    assert request.get_header('Cache-control') == 'no-cache'
    assert request.get_header('Accept') == 'application/json'
    assert '/release.json?check=' in request.full_url
    output = capsys.readouterr().out
    assert 'HTTP 503' in output
    assert 'backend: sha=' + SHA + ', build_id=' + 'c' * 32 in output
    assert 'Verified deployed release' in output
    assert 'secret.example' not in output


@pytest.mark.parametrize('failure, expected', [
    (lambda: HTTPError('https://secret.example', 403, 'secret response', {}, None), 'HTTP 403'),
    (lambda: URLError(OSError('secret address')), 'connection failed (OSError)'),
    (lambda: TimeoutError('secret address'), 'request failed (TimeoutError)'),
    (lambda: BytesIO(b'<html>secret response</html>'), 'invalid JSON'),
    (lambda: response([]), 'expected a JSON object'),
    (lambda: response({'frontend': {'sha': 'secret response'}}), 'missing or invalid'),
])
def test_wait_timeout_reports_last_failure_without_leaking_response_or_url(poll, capsys, failure, expected):
    poll.side_effect = lambda *args, **kwargs: result_or_raise(failure())
    with pytest.raises(RuntimeError) as error:
        release.wait('https://secret.example', SHA, BUILD, 10)
    assert expected in str(error.value)
    assert SHA in str(error.value)
    assert BUILD in str(error.value)
    output = capsys.readouterr().out
    assert sum(expected in line for line in output.splitlines()) == 1
    assert 'secret' not in output + str(error.value)
    assert poll.call_count == 2


def result_or_raise(result):
    if isinstance(result, Exception):
        raise result
    return result