coulomb-social/tests/test_bubble_visuals.py

"""Synthetic visual capture: origin limits, immutable source, complete private output."""

import hashlib
import importlib.util
import json
from pathlib import Path

import pytest

from coulomb_social.apps.spaces.bubble_bundle import prepare_bundle
from tests.test_bubble_bundle import capture as source_capture

SPEC = importlib.util.spec_from_file_location(
    'bubble_visuals', Path(__file__).resolve().parents[1] / 'scripts/capture-bubble-visuals.py')
visuals = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(visuals)
URL = 'https://' + visuals.CDN + '/synthetic.png'
PNG = b'\x89PNG\r\n\x1a\nsynthetic-no-decode-claim'


def bundle(tmp_path):
    tmp_path.chmod(0o700)
    source = tmp_path / 'source'
    source.mkdir(mode=0o700)
    spaces, articles = source_capture()
    articles['r']['logo_image'] = '//' + visuals.CDN + '/synthetic.png'
    articles['c']['logo_image'] = URL
    # Preserve the blocked and unassigned records as well as all eligible draft pages.
    articles['orphan'] = {'logo_image': URL, 'content_text': 'Unassigned synthetic body'}
    report, files = prepare_bundle(spaces, articles, inventory_spaces={'missing': {}})
    inputs = {k: json.dumps(v).encode() for k, v in [('spaces', spaces), ('articles', articles)]}
    report['source_sha256'] = {k: hashlib.sha256(v).hexdigest() for k, v in inputs.items()}
    report['status'] = 'completed_offline_preparation'
    for name, raw in {**files, **{'source-capture/' + k + '.json': v
                                 for k, v in inputs.items()}}.items():
        visuals.publish(source, name, raw)
    visuals.publish(source, 'reconciliation.json', json.dumps(report).encode())
    return source, report, files


def test_shared_visuals_private_bundle_preserves_bodies_dates_gates_and_source(tmp_path):
    source, report, files = bundle(tmp_path)
    before = {str(p.relative_to(source)): p.read_bytes() for p in source.rglob('*') if p.is_file()}
    calls = []
    def fetch(url):
        calls.append(url)
        return PNG, 'png'
    out = tmp_path / 'output'
    result = visuals.capture(source, out, fetch)
    assert calls == [URL]
    assert result['references'] == 3 and result['unique_urls'] == 1
    assert result['materialized_pages'] == 2
    after = json.loads((out / 'reconciliation.json').read_bytes())
    assert after['unassigned_article_ids'] == report['unassigned_article_ids']
    assert len(after['spaces']) == len(report['spaces'])
    for original, updated in zip(report['spaces'], after['spaces'], strict=True):
        assert updated['blockers'] == original['blockers'] and not updated['import_ready']
        for page in updated['articles']:
            if page['file']:
                raw = (out / page['file']).read_bytes()
                assert raw.split(b'\n---\n', 1)[1] == files[page['file']].split(b'\n---\n', 1)[1]
                assert b'source_created_at:' in raw and b'visual: "assets/' in raw
    for name, digest in after['file_sha256'].items():
        assert visuals.sha((out / name).read_bytes()) == digest
    assert before == {str(p.relative_to(source)): p.read_bytes()
                      for p in source.rglob('*') if p.is_file()}
    for p in out.rglob('*'):
        assert p.stat().st_mode & 0o077 == 0
    with pytest.raises(FileExistsError):
        visuals.capture(source, out, fetch)
    assert calls == [URL]


@pytest.mark.parametrize('value', ['http://' + visuals.CDN + '/x',
    'https://example.invalid/x', 'https://user@' + visuals.CDN + '/x',
    'https://' + visuals.CDN + ':8443/x', 'https://' + visuals.CDN + '/x?q=secret',
    'https://' + visuals.CDN + '/x#fragment', 'https://' + visuals.CDN + '/x\n',
    'https://' + visuals.CDN + '.example.invalid/x', '//127.0.0.1/x'])
def test_unreviewed_urls_refused(value):
    with pytest.raises(visuals.Refused):
        visuals.canonical_url(value)


def test_redirects_and_active_content_refused():
    with pytest.raises(visuals.Refused):
        visuals.NoRedirect().redirect_request(None, None, 302, '', {}, URL)
    for raw, kind in [(b'<svg/>', 'image/svg+xml'), (PNG, 'text/html'), (b'html', 'image/png')]:
        with pytest.raises(visuals.Refused):
            visuals.image_extension(raw, kind)


@pytest.mark.parametrize('defect', ['file-drift', 'outside-tree', 'symlink', 'public-source'])
def test_bad_source_refuses_before_fetch_or_output(tmp_path, defect):
    source, report, _ = bundle(tmp_path)
    name = next(iter(report['file_sha256']))
    if defect == 'file-drift':
        (source / name).write_bytes(b'changed')
    elif defect == 'outside-tree':
        report['spaces'][0]['tree'] = '/outside'
        (source / 'reconciliation.json').write_text(json.dumps(report))
    elif defect == 'symlink':
        (source / name).unlink()
        (source / name).symlink_to(tmp_path / 'missing')
    else:
        (source / name).chmod(0o644)
    out = tmp_path / 'out'
    with pytest.raises((visuals.Refused, KeyError)):
        visuals.capture(source, out, lambda u: pytest.fail('network called'))
    assert not out.exists()


@pytest.mark.parametrize('defect', ['fetch-failure', 'during-fetch-drift', 'budget'])
def test_partial_capture_never_publishes_completion(tmp_path, monkeypatch, defect):
    source, report, _ = bundle(tmp_path)
    if defect == 'budget':
        monkeypatch.setattr(visuals, 'MAX_TOTAL', 1)
    def fetch(url):
        if defect == 'fetch-failure':
            raise RuntimeError('private-url-do-not-print')
        if defect == 'during-fetch-drift':
            (source / next(iter(report['file_sha256']))).write_bytes(b'changed')
        return PNG, 'png'
    out = tmp_path / 'out'
    with pytest.raises((visuals.Refused, RuntimeError)):
        visuals.capture(source, out, fetch)
    assert not (out / 'reconciliation.json').exists()


@pytest.mark.parametrize('defect', ['status', 'redirected-url', 'oversize-header',
                                   'oversize-body', 'short-body', 'active-type'])
def test_fetch_rejects_invalid_transport_results_without_followup(monkeypatch, defect):
    from email.message import Message
    class Response:
        status = 404 if defect == 'status' else 200
        headers = Message()
        headers['Content-Type'] = 'text/html' if defect == 'active-type' else 'image/png'
        headers['Content-Length'] = str(
            visuals.MAX_ASSET + 1 if defect == 'oversize-header' else len(PNG) + 1
            if defect == 'short-body' else len(PNG))
        def __enter__(self):
            return self
        def __exit__(self, *args):
            pass
        def geturl(self):
            return 'https://example.invalid/x' if defect == 'redirected-url' else URL
        def read(self, maximum):
            assert maximum == visuals.MAX_ASSET + 1
            return b'x' * maximum if defect == 'oversize-body' else PNG
    class Opener:
        def open(self, request, timeout):
            assert request.full_url == URL and timeout == 20
            assert not any(k.lower() in {'authorization', 'cookie', 'referer'}
                           for k in request.headers)
            return Response()
    monkeypatch.setattr(visuals.urllib.request, 'build_opener', lambda *args: Opener())
    with pytest.raises(visuals.Refused):
        visuals.fetch(URL)