bmc_hub/tests/test_vendor_resources.py

import asyncio
from io import BytesIO
from types import SimpleNamespace

import pytest
from fastapi import HTTPException, UploadFile
from app.vendors.backend import resources as r


def run(call):
    return asyncio.run(call)


def test_unknown_vendor_blocks_queries(monkeypatch):
    monkeypatch.setattr(r, 'execute_query_single', lambda *args: None)
    with pytest.raises(HTTPException) as error:
        run(r.files(999))
    assert error.value.status_code == 404


def test_upload_limits_and_safe_filename(monkeypatch):
    saved = []
    def query(sql, params):
        if sql.startswith('SELECT'): return {'id': 1}
        saved.append(params)
        return {'id': 2}
    monkeypatch.setattr(r, 'execute_query_single', query)
    request = SimpleNamespace(state=SimpleNamespace(user_id=3))
    run(r.upload_file(1, request, UploadFile(filename='../test.pdf', file=BytesIO(b'abc'))))
    assert saved[0][1] == 'test.pdf'
    assert saved[0][3] == 3
    assert saved[0][-1] == 3
    monkeypatch.setattr(r, 'MAX_FILE_BYTES', 2)
    with pytest.raises(HTTPException) as error:
        run(r.upload_file(1, request, UploadFile(filename='big', file=BytesIO(b'abc'))))
    assert error.value.status_code == 413
    assert len(saved) == 1


def test_db_only_attachment_download_is_vendor_scoped(monkeypatch):
    captured = []
    def query(sql, params):
        if sql.startswith('SELECT id FROM vendors'): return {'id': 1}
        captured.append((sql, params))
        return {'filename': 'bilag.pdf', 'content_data': memoryview(b'pdf'), 'file_path': None}
    monkeypatch.setattr(r, 'execute_query_single', query)
    result = run(r.download_file(1, 'email', 7))
    assert result.body == b'pdf'
    assert captured[0][1] == (1, 7)
    assert 'e.supplier_id=%s' in captured[0][0]
    assert result.headers['content-disposition'].startswith('attachment;')


def test_missing_foreign_attachment_is_not_downloaded(monkeypatch):
    monkeypatch.setattr(r, 'execute_query_single', lambda sql, params: {'id':1} if sql.startswith('SELECT id FROM vendors') else None)
    with pytest.raises(HTTPException) as error:
        run(r.download_file(1, 'email', 8))
    assert error.value.status_code == 404


def test_email_preview_is_pure_and_sanitized(monkeypatch):
    monkeypatch.setattr(r, 'execute_query_single', lambda sql, params: {'id':1} if sql.startswith('SELECT id FROM vendors') else {'body_html':'<script>bad()</script><p>Mail</p>'})
    monkeypatch.setattr(r, 'execute_update', lambda *args: pytest.fail('preview must not write'))
    result = run(r.email_detail(1, 7))
    assert result['body_html'] == '<p>Mail</p>'


def test_cases_do_not_include_every_customer_case(monkeypatch):
    monkeypatch.setattr(r, 'execute_query_single', lambda *args: {'id':1})
    seen = []
    monkeypatch.setattr(r, 'execute_query', lambda sql, params: seen.append((sql,params)) or [])
    run(r.cases(1))
    assert 'customer_vendor_links' not in seen[0][0]
    assert 'sag_emails' in seen[0][0]
    assert seen[0][1] == (1,1,1)


def test_invalid_case_never_creates_link(monkeypatch):
    monkeypatch.setattr(r, 'execute_query_single', lambda sql, params: {'id':1} if sql.startswith('SELECT id FROM vendors') else None)
    monkeypatch.setattr(r, 'execute_query', lambda *args: pytest.fail('no writes on invalid case'))
    with pytest.raises(HTTPException): run(r.link_case(1, r.CaseLink(sag_id=9)))


def test_contact_link_keeps_existing_contact_and_scopes_removal(monkeypatch):
    monkeypatch.setattr(r, 'execute_query_single', lambda *args: {'id':1})
    seen = []
    monkeypatch.setattr(r, 'execute_query', lambda sql, params: seen.append((sql,params)))
    monkeypatch.setattr(r, 'execute_update', lambda sql, params: seen.append((sql,params)))
    run(r.link_contact(1,r.ContactLink(contact_id=3,role=' Support ')))
    run(r.unlink_contact(1,3))
    assert seen[0][1] == (1,3,'Support',None,None)
    assert seen[1][1] == (1,3)
    assert 'DELETE FROM vendor_contacts' in seen[1][0]


def test_pdf_can_open_inline_but_default_is_download():
    content = b'%PDF-1.7\nfixture'
    opened = r.download(content, 'bilag.pdf', inline=True)
    saved = r.download(content, 'bilag.pdf')
    assert opened.headers['content-type'] == 'application/pdf'
    assert opened.headers['content-disposition'].startswith('inline;')
    assert saved.headers['content-disposition'].startswith('attachment;')


def test_active_html_is_never_opened_inline():
    response = r.download(b'<script>bad()</script>', 'fil.html', inline=True)
    assert response.headers['content-type'] == 'application/octet-stream'
    assert response.headers['content-disposition'].startswith('attachment;')


def test_text_opens_as_plain_text():
    response = r.download(b'<script>text</script>', 'noter.txt', inline=True)
    assert response.headers['content-type'].startswith('text/plain')
    assert response.headers['content-disposition'].startswith('inline;')


def test_contact_primary_is_optional_and_explicit(monkeypatch):
    monkeypatch.setattr(r, 'execute_query_single', lambda *args: {'id':1})
    seen=[]
    monkeypatch.setattr(r,'execute_query',lambda sql,params:seen.append((sql,params)))
    run(r.link_contact(1,r.ContactLink(contact_id=3,is_primary=True)))
    assert seen[0][1][-2:]==(True,True)
    assert 'COALESCE(%s,vendor_contacts.is_primary)' in seen[0][0]