"""
tests/test_storage_seam_guard.py
--------------------------------
MT-22: a static guard against re-opening the storage seam.

Every uploaded file must go through ``app/utils/storage.py``. Any code that
writes bytes to disk directly survives the R2 cutover as a local-only file:
the tenant's other media moves to the bucket, that one file does not, and on
the local backend it lands in a directory shared by every tenant with no
``t<id>/`` prefix.

That is exactly how ``tenant_settings._save_logo`` stayed broken — it was
written before the seam existed and nothing failed when the seam arrived. A
runtime test cannot catch this class of bug, because a direct write *works*;
it just works in the wrong place. So this scans the source instead.

Adding a new writer is a deliberate act: if one is genuinely needed (a cache,
a temp file), mark the line ``# storage-seam-exempt: <reason>``.
"""

import ast
import pathlib

import pytest

APP_ROOT = pathlib.Path(__file__).resolve().parent.parent / 'app'

EXEMPT_MARKER = 'storage-seam-exempt'

# storage.py is the seam itself; photo_stamp works on already-materialized
# temp files handed to it by the seam's materialize_to_dir().
EXEMPT_FILES = {
    'utils/storage.py',
    'utils/photo_stamp.py',
}


def _python_files():
    for path in sorted(APP_ROOT.rglob('*.py')):
        rel = path.relative_to(APP_ROOT).as_posix()
        if rel in EXEMPT_FILES:
            continue
        yield rel, path


def _is_exempt_line(source_lines, lineno):
    """True when the flagged line (or the one above it) carries the marker."""
    for idx in (lineno - 1, lineno - 2):
        if 0 <= idx < len(source_lines) and EXEMPT_MARKER in source_lines[idx]:
            return True
    return False


def test_no_direct_uploaded_file_saves():
    """No route or util may call ``<werkzeug FileStorage>.save(...)``.

    Matches any ``*.save(path)`` where the receiver name looks like an upload
    (``file``, ``file_obj``, ``photo``, ``logo``, ``upload``...). ORM
    ``db.session.save`` and friends take no positional path, so they don't
    match the shape here.
    """
    offenders = []
    upload_names = ('file', 'photo', 'image', 'logo', 'upload', 'attachment')

    for rel, path in _python_files():
        source = path.read_text(encoding='utf-8-sig')
        lines = source.splitlines()
        try:
            tree = ast.parse(source)
        except SyntaxError as exc:                      # pragma: no cover
            pytest.fail(f'{rel} does not parse: {exc}')

        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            if not (isinstance(func, ast.Attribute) and func.attr == 'save'):
                continue
            if not node.args:                           # .save() with no path
                continue
            receiver = func.value
            name = getattr(receiver, 'id', None) or getattr(receiver, 'attr', '')
            if not any(token in name.lower() for token in upload_names):
                continue
            if _is_exempt_line(lines, node.lineno):
                continue
            offenders.append(f'{rel}:{node.lineno} — {name}.save(...)')

    assert not offenders, (
        'Direct file writes bypass the storage seam (app/utils/storage.py). '
        'Use storage.save(file_obj, subfolder) instead, or mark the line '
        f'"# {EXEMPT_MARKER}: <reason>".\n  ' + '\n  '.join(offenders)
    )


def test_no_direct_writes_under_upload_folder():
    """Nothing may build a path from ``UPLOAD_FOLDER`` and write to it.

    The seam owns that directory. Reading config for display or for a
    migration script is fine; opening a file under it for writing is not.
    """
    offenders = []

    for rel, path in _python_files():
        lines = path.read_text(encoding='utf-8-sig').splitlines()
        for i, line in enumerate(lines, start=1):
            if 'UPLOAD_FOLDER' not in line:
                continue
            if _is_exempt_line(lines, i):
                continue
            # Creating the uploads root itself at startup is the seam's own
            # precondition. What matters is code reaching *into* it — that
            # always joins a subfolder or opens a file.
            if 'os.path.join' in line or 'open(' in line:
                offenders.append(f'{rel}:{i} — {line.strip()}')

    assert not offenders, (
        'Direct writes under UPLOAD_FOLDER bypass the storage seam.\n  '
        + '\n  '.join(offenders)
    )


def test_logo_helpers_use_the_seam():
    """Pin the MT-22 fix itself so it cannot silently regress."""
    source = (APP_ROOT / 'routes' / 'tenant_settings.py').read_text(encoding='utf-8-sig')
    assert 'from app.utils import storage' in source, \
        'tenant_settings.py no longer imports the storage seam'
    assert "storage.save(file_obj, 'logos')" in source, \
        '_save_logo() no longer writes through storage.save()'
    assert 'storage.delete(' in source, \
        '_delete_logo() no longer deletes through the storage seam'


def test_logo_templates_use_media_url():
    """Logos must render through media_url(), not url_for('static', ...).

    On an R2-backed tenant a static URL points at a file that isn't there.
    """
    templates = [
        APP_ROOT / 'templates' / 'layouts' / 'modern.html',
        APP_ROOT / 'templates' / 'layouts' / 'classic.html',
        APP_ROOT / 'templates' / 'tenant_settings' / 'branding.html',
    ]
    offenders = []
    for tpl in templates:
        for i, line in enumerate(tpl.read_text(encoding='utf-8-sig').splitlines(), start=1):
            if 'logo_url' in line and 'url_for(' in line and 'static' in line:
                offenders.append(f'{tpl.name}:{i} — {line.strip()}')

    assert not offenders, (
        'Logo rendered via url_for(static) instead of media_url().\n  '
        + '\n  '.join(offenders)
    )


# ── Functional: the logo path now behaves like every other upload ────────────
#
# The static guards above prove no direct writer remains. These prove the
# replacement is correct: tenant-prefixed on the wire, bare key in the DB,
# and rejected outright when no tenant is bound.

import io

from flask import g

from app.utils import storage
from app.routes.tenant_settings import _save_logo, _delete_logo


class _FakeTenant:
    def __init__(self, tid):
        self.id = tid


class _FakeUpload:
    """Stands in for a werkzeug FileStorage well enough for _save_logo()."""

    def __init__(self, filename='logo.png', data=b'\x89PNG\r\n\x1a\n' + b'x' * 32):
        self.filename = filename
        self.stream = io.BytesIO(data)

    # _save_logo reads magic bytes off the object itself before handing it on.
    def read(self, n=-1):
        return self.stream.read(n)

    def seek(self, pos):
        return self.stream.seek(pos)

    def save(self, dest):                      # storage-seam-exempt: test double
        raise AssertionError(
            '_save_logo() wrote directly to disk instead of using storage.save()'
        )


class _StubClient:
    def __init__(self):
        self.put_calls = []
        self.deleted = []

    def put_object(self, Bucket=None, Key=None, Body=None, ContentType=None):
        self.put_calls.append((Key, ContentType))
        return {}

    def head_object(self, Bucket=None, Key=None):
        raise RuntimeError('404')

    def delete_object(self, Bucket=None, Key=None):
        self.deleted.append(Key)
        return {}


def _install_s3(app, stub):
    backend = object.__new__(storage.S3Backend)
    backend.bucket = 'test-bucket'
    backend.ttl = 60
    backend.fallback = False
    backend._client = stub
    backend._local = storage.LocalBackend()
    app.extensions['_storage_backend_s3'] = backend
    app.config['STORAGE_BACKEND'] = 's3'
    return backend


@pytest.fixture
def mt_s3(app):
    """App context: multi-tenancy on, tenant 7 bound, stubbed S3 backend."""
    stub = _StubClient()
    with app.test_request_context('/'):
        app.config['MULTI_TENANT_ENABLED'] = True
        g.tenant = _FakeTenant(7)
        _install_s3(app, stub)
        try:
            yield stub
        finally:
            app.config['MULTI_TENANT_ENABLED'] = False
            app.config['STORAGE_BACKEND'] = 'local'
            app.extensions.pop('_storage_backend_s3', None)


def test_logo_save_is_tenant_prefixed_on_the_wire(mt_s3):
    key = _save_logo(_FakeUpload('Company Logo.PNG'))

    # DB value keeps the historic shape — no migration, no data rewrite.
    assert key.startswith('uploads/logos/')
    assert key.endswith('.png')
    assert not key.startswith('t7/')
    # The object itself is scoped to the tenant.
    assert mt_s3.put_calls == [(f't7/{key}', 'image/png')]


def test_logo_delete_targets_the_prefixed_object(mt_s3):
    _delete_logo('uploads/logos/abc123.png')
    assert 't7/uploads/logos/abc123.png' in mt_s3.deleted


def test_logo_delete_ignores_empty_value(mt_s3):
    _delete_logo(None)
    _delete_logo('')
    assert mt_s3.deleted == []


def test_logo_delete_cannot_escape_the_logos_folder(mt_s3):
    """A tampered column value must not reach another subfolder's objects.

    S3Backend.delete() probes both the tenant-prefixed key and the legacy
    unprefixed one (its pre-existing transition behaviour, shared by all
    media), so assert containment in uploads/logos/ rather than the prefix.
    """
    _delete_logo('uploads/issue_photos/../../secret.jpg')
    assert mt_s3.deleted, 'expected a delete attempt'
    for key in mt_s3.deleted:
        assert 'issue_photos' not in key
        assert key.endswith('uploads/logos/secret.jpg')


def test_logo_save_refuses_with_no_tenant_bound(app):
    """Same loud failure as any other upload — never an unprefixed key."""
    stub = _StubClient()
    with app.test_request_context('/'):
        app.config['MULTI_TENANT_ENABLED'] = True
        g.tenant = None
        _install_s3(app, stub)
        try:
            with pytest.raises(RuntimeError):
                _save_logo(_FakeUpload())
        finally:
            app.config['MULTI_TENANT_ENABLED'] = False
            app.config['STORAGE_BACKEND'] = 'local'
            app.extensions.pop('_storage_backend_s3', None)
    assert stub.put_calls == []


def test_logo_validation_still_rejects_non_images(mt_s3):
    """Validation stays in the caller — the seam only moves bytes."""
    assert _save_logo(_FakeUpload('payload.exe', b'MZ\x90\x00')) is None
    assert _save_logo(_FakeUpload('fake.png', b'not-an-image')) is None
    assert _save_logo(None) is None
    assert mt_s3.put_calls == []


def test_svg_logo_gets_a_renderable_content_type(mt_s3):
    """SVG skips the magic-byte check; it must still upload as an image type."""
    key = _save_logo(_FakeUpload('brand.svg', b'<svg xmlns="..."></svg>'))
    assert key.endswith('.svg')
    assert mt_s3.put_calls == [(f't7/{key}', 'image/svg+xml')]