diff --git a/plugins/warranty/api/routes.py b/plugins/warranty/api/routes.py index 85fd0da..11011a4 100644 --- a/plugins/warranty/api/routes.py +++ b/plugins/warranty/api/routes.py @@ -5,11 +5,14 @@ never stored. Warranties link to assets many-to-many via warrantyassets, though the common case is one warranty per asset. """ +import glob +import os from datetime import date, datetime, timedelta, timezone -from flask import Blueprint, request +from flask import Blueprint, request, current_app, send_from_directory from flask_jwt_extended import jwt_required from sqlalchemy.orm import joinedload +from werkzeug.utils import secure_filename from shopdb.api import ( db, Asset, @@ -22,6 +25,21 @@ from ..services import get_provider, ProviderNotConfigured, WarrantyLookupError warranty_bp = Blueprint('warranty', __name__) +# What a proof of cover actually arrives as: a vendor PDF, a scan, or a +# screenshot of a portal page. Office formats are allowed because purchase +# records often arrive that way. +PROOF_EXTENSIONS = {'.pdf', '.png', '.jpg', '.jpeg', '.gif', '.webp', '.tif', + '.tiff', '.msg', '.eml', '.doc', '.docx', '.xls', '.xlsx'} +PROOF_URL_PREFIX = '/api/warranty/proof/' + +# A certificate is a document, not a disk image. Anything past this is somebody +# attaching the wrong thing. +MAX_PROOF_BYTES = 25 * 1024 * 1024 + + +def _proof_dir(): + return os.path.join(current_app.instance_path, 'warrantyproofs') + def _parse_date(value): """Accept 'YYYY-MM-DD' (or None/empty) -> date or None.""" @@ -466,3 +484,100 @@ def _hostname(asset): return None computer = Computer.query.filter_by(assetid=asset.assetid).first() return computer.hostname if computer else None + + +# ============================================================================= +# Proof of cover +# +# Authenticated on the way in AND on the way out: an invoice carries prices and +# a service tag, so it is not something to serve openly the way an asset photo +# is. Stored as warranty-, one per warranty, with the vendor's own +# filename kept alongside so a person recognises it later. +# ============================================================================= + +@warranty_bp.route('//proof', methods=['POST']) +@jwt_required() +@require_permission('warranty.edit') +def upload_proof(warrantyid): + """Upload (or replace) the proof-of-cover document for a warranty.""" + warranty = db.session.get(Warranty, warrantyid) + if not warranty: + return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', + http_code=404) + + upload = request.files.get('file') + if not upload or not upload.filename: + return error_response(ErrorCodes.VALIDATION_ERROR, 'No file provided') + + ext = os.path.splitext(upload.filename)[1].lower() + if ext not in PROOF_EXTENSIONS: + return error_response( + ErrorCodes.VALIDATION_ERROR, + 'Unsupported document type {}. Allowed: {}'.format( + ext, ', '.join(sorted(PROOF_EXTENSIONS)))) + + # Seek rather than trust Content-Length: a chunked upload has no length + # header, and a client can understate the one it sends. + upload.stream.seek(0, os.SEEK_END) + size = upload.stream.tell() + upload.stream.seek(0) + if size > MAX_PROOF_BYTES: + return error_response( + ErrorCodes.VALIDATION_ERROR, + 'Document is {:.0f}MB; the limit is {:.0f}MB.'.format( + size / 1048576, MAX_PROOF_BYTES / 1048576)) + + proofdir = _proof_dir() + os.makedirs(proofdir, exist_ok=True) + # One proof per warranty: clear any prior extension so a re-upload as .pdf + # does not leave the old .png behind claiming to be current. + for old in glob.glob(os.path.join(proofdir, + secure_filename(f'warranty-{warrantyid}') + '.*')): + os.remove(old) + + filename = secure_filename(f'warranty-{warrantyid}{ext}') + upload.save(os.path.join(proofdir, filename)) + + warranty.proofurl = f'{PROOF_URL_PREFIX}{filename}' + warranty.prooffilename = upload.filename + db.session.commit() + + return success_response(warranty.to_dict(), message='Proof uploaded') + + +@warranty_bp.route('/proof/', methods=['GET']) +@jwt_required() +@require_permission('warranty.view') +def serve_proof(filename): + """Download a proof document. + + Authenticated: an invoice carries pricing and a service tag. Sent as an + attachment under the vendor's original filename where we still have it, so + a download is recognisable rather than 'warranty-12.pdf'. + """ + warranty = Warranty.query.filter( + Warranty.proofurl == f'{PROOF_URL_PREFIX}{filename}').first() + downloadname = (warranty.prooffilename if warranty and warranty.prooffilename + else filename) + return send_from_directory(_proof_dir(), filename, as_attachment=True, + download_name=downloadname) + + +@warranty_bp.route('//proof', methods=['DELETE']) +@jwt_required() +@require_permission('warranty.edit') +def delete_proof(warrantyid): + """Remove a warranty's proof document.""" + warranty = db.session.get(Warranty, warrantyid) + if not warranty: + return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', + http_code=404) + + for old in glob.glob(os.path.join(_proof_dir(), + secure_filename(f'warranty-{warrantyid}') + '.*')): + os.remove(old) + warranty.proofurl = None + warranty.prooffilename = None + db.session.commit() + + return success_response(warranty.to_dict(), message='Proof removed') diff --git a/plugins/warranty/frontend/views/WarrantiesList.vue b/plugins/warranty/frontend/views/WarrantiesList.vue index fdbf34e..e208cc0 100644 --- a/plugins/warranty/frontend/views/WarrantiesList.vue +++ b/plugins/warranty/frontend/views/WarrantiesList.vue @@ -151,6 +151,21 @@ + +
+ +
+ {{ proofName }} + {{ proofName }} + +
+ + + Invoice, certificate or a saved email - up to 25MB. Downloading + needs a login, since it carries pricing and a service tag. + +
{{ error }}
@@ -167,7 +182,7 @@ import { ref, computed, watch, onMounted } from 'vue' import { useRoute } from 'vue-router' import { colorStyle } from '@/utils/colorStyle' -import { warrantyApi, assetsApi, vendorsApi } from '@/api' +import api, { warrantyApi, assetsApi, vendorsApi } from '@/api' import { useToast } from '@/composables/toast' import { apiError } from '@/utils/apiError' import MachineMapChip from '../components/MachineMapChip.vue' @@ -315,7 +330,10 @@ function openModal(item = null) { } else { form.value = blankForm() selectedAssets.value = [] + proofName.value = '' + proofUrl.value = '' } + pendingProof.value = null assetQuery.value = '' assetResults.value = [] error.value = '' @@ -323,15 +341,75 @@ function openModal(item = null) { } function closeModal() { showModal.value = false; editing.value = null } +// Chosen before a new warranty exists, so it is held here and uploaded once +// there is an id to attach it to. +const pendingProof = ref(null) +const proofName = ref('') +const proofUrl = ref('') +const busyProof = ref(false) +const PROOF_ACCEPT = '.pdf,.png,.jpg,.jpeg,.gif,.webp,.tif,.tiff,.msg,.eml,.doc,.docx,.xls,.xlsx' +const MAX_PROOF_BYTES = 25 * 1024 * 1024 + +function onProofPicked(event) { + const file = event.target.files?.[0] + if (!file) return + if (file.size > MAX_PROOF_BYTES) { + error.value = `${file.name} is ${(file.size / 1048576).toFixed(0)}MB; the limit is 25MB.` + event.target.value = '' + return + } + error.value = '' + pendingProof.value = file + proofName.value = file.name + proofUrl.value = '' +} + +async function removeProof() { + pendingProof.value = null + proofName.value = '' + proofUrl.value = '' + if (editing.value) { + busyProof.value = true + try { await warrantyApi.removeProof(editing.value.warrantyid) } + catch (err) { error.value = apiError(err, 'Failed to remove the document') } + finally { busyProof.value = false } + } +} + +// The route is authenticated, so a plain href would 401. Fetch it through the +// client, which carries the token, then hand the blob to the browser. +async function downloadProof() { + try { + const response = await api.get(proofUrl.value, { responseType: 'blob' }) + const url = URL.createObjectURL(response.data) + const link = document.createElement('a') + link.href = url + link.download = proofName.value || 'proof' + link.click() + URL.revokeObjectURL(url) + } catch (err) { + error.value = apiError(err, 'Failed to download the document') + } +} + async function save() { error.value = '' saving.value = true try { const payload = { ...form.value, assetids: selectedAssets.value.map(a => a.assetid) } + let warrantyid if (editing.value) { - await warrantyApi.update(editing.value.warrantyid, payload) + warrantyid = editing.value.warrantyid + await warrantyApi.update(warrantyid, payload) } else { - await warrantyApi.create(payload) + const created = await warrantyApi.create(payload) + warrantyid = created.data.data.warrantyid + } + // After the save: a new warranty has no id until it exists, and the upload + // keys on it. A failed upload leaves the saved record alone and reports. + if (pendingProof.value) { + await warrantyApi.uploadProof(warrantyid, pendingProof.value) + pendingProof.value = null } closeModal() loadData() @@ -390,6 +468,8 @@ async function refresh(w) { .filters label { display: inline-flex; align-items: center; gap: 0.4rem; } .filters .form-control { max-width: 320px; } .result-count { color: var(--text-light); font-size: 0.85rem; } +.proof-current { display: flex; align-items: center; gap: 0.75rem; margin-bottom: 0.5rem; } +.proof-current a { word-break: break-all; } .servicelevel-cell { max-width: 320px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } .pagination { display: flex; align-items: center; justify-content: center; gap: 1rem; padding: 0.9rem 0 0.2rem; } .page-info { color: var(--text-light); font-size: 0.85rem; } diff --git a/plugins/warranty/migrations/versions/0002_proof_document.py b/plugins/warranty/migrations/versions/0002_proof_document.py new file mode 100644 index 0000000..3723390 --- /dev/null +++ b/plugins/warranty/migrations/versions/0002_proof_document.py @@ -0,0 +1,50 @@ +"""Add warranties.proofurl / prooffilename (proof-of-cover document). + +A provider lookup answers whether a unit is covered. It does not produce the +invoice or the extended-warranty certificate, and a manually entered warranty +had nowhere to keep one - so the proof lived in somebody's mailbox until they +left. Two columns: the served URL of the stored file, and the name the vendor +sent it under, which is what a person recognises months later. + +Idempotent; downgrade drops both. + +Revision ID: warranty0002proof +Revises: warranty0001anchor +""" +from alembic import op +import sqlalchemy as sa + + +revision = 'warranty0002proof' +down_revision = 'warranty0001anchor' +branch_labels = None +depends_on = None + +_TABLE = 'warranties' +_COLUMNS = (('proofurl', sa.String(500)), ('prooffilename', sa.String(255))) + + +def _column_names(insp, table): + return {c['name'] for c in insp.get_columns(table)} + + +def upgrade(): + bind = op.get_bind() + insp = sa.inspect(bind) + if _TABLE not in insp.get_table_names(): + return + existing = _column_names(insp, _TABLE) + for name, coltype in _COLUMNS: + if name not in existing: + op.add_column(_TABLE, sa.Column(name, coltype, nullable=True)) + + +def downgrade(): + bind = op.get_bind() + insp = sa.inspect(bind) + if _TABLE not in insp.get_table_names(): + return + existing = _column_names(insp, _TABLE) + for name, _ in _COLUMNS: + if name in existing: + op.drop_column(_TABLE, name) diff --git a/plugins/warranty/models/warranty.py b/plugins/warranty/models/warranty.py index 8199620..cd4823e 100644 --- a/plugins/warranty/models/warranty.py +++ b/plugins/warranty/models/warranty.py @@ -48,6 +48,14 @@ class Warranty(db.Model): # When a provider lookup last refreshed this record. lastcheckeddate = db.Column(db.DateTime) notes = db.Column(db.Text) + + # Proof of cover: the purchase invoice, the extended-warranty certificate, + # whatever the vendor sent. A provider lookup answers "is it covered"; this + # answers "prove it" months later, when the email it arrived in is gone. + # Stores the served URL, not the original filename - see the upload route. + proofurl = db.Column(db.String(500)) + prooffilename = db.Column(db.String(255)) + isactive = db.Column(db.Boolean, nullable=False, server_default='1') links = db.relationship('WarrantyAsset', back_populates='warranty', @@ -68,6 +76,8 @@ class Warranty(db.Model): 'enddate': self.enddate.isoformat() if self.enddate else None, 'lastcheckeddate': self.lastcheckeddate.isoformat() + 'Z' if self.lastcheckeddate else None, 'notes': self.notes, + 'proofurl': self.proofurl, + 'prooffilename': self.prooffilename, 'isactive': bool(self.isactive), 'status': status, 'statuscolor': STATUS_COLORS.get(status, STATUS_COLORS['unknown']), diff --git a/tests/test_plugins/test_warranty_proof.py b/tests/test_plugins/test_warranty_proof.py new file mode 100644 index 0000000..c0c4218 --- /dev/null +++ b/tests/test_plugins/test_warranty_proof.py @@ -0,0 +1,110 @@ +"""Tests for the warranty proof-of-cover document. + +A provider lookup answers whether a unit is covered; it does not produce the +invoice or the extended-warranty certificate. Those had nowhere to live, so +they stayed in somebody's mailbox. These pin the parts that are easy to get +wrong: the document is authenticated on the way out, one per warranty, and the +vendor's own filename survives so a download is recognisable. +""" + +import io + +from plugins.warranty.models import Warranty + + +def _warranty(db, vendor='Dell'): + w = Warranty(vendor=vendor, servicetag='ABC123', provider='manual') + db.session.add(w) + db.session.commit() + return w + + +def test_upload_sets_url_and_keeps_the_original_filename(client, db, auth_headers): + w = _warranty(db) + + response = client.post(f'/api/warranty/{w.warrantyid}/proof', + data={'file': (io.BytesIO(b'%PDF-1.4 invoice'), + 'Dell invoice 4471.pdf')}, + content_type='multipart/form-data', + headers=auth_headers) + + assert response.status_code == 200, response.get_json() + body = response.get_json()['data'] + assert body['proofurl'] == f'/api/warranty/proof/warranty-{w.warrantyid}.pdf' + assert body['prooffilename'] == 'Dell invoice 4471.pdf' + + +def test_download_needs_a_token_and_uses_the_vendor_filename(client, db, auth_headers): + """An invoice carries pricing and a service tag - not open like an asset photo.""" + w = _warranty(db) + client.post(f'/api/warranty/{w.warrantyid}/proof', + data={'file': (io.BytesIO(b'%PDF-1.4 invoice'), 'Dell invoice 4471.pdf')}, + content_type='multipart/form-data', headers=auth_headers) + path = f'/api/warranty/proof/warranty-{w.warrantyid}.pdf' + + assert client.get(path).status_code == 401 + + authed = client.get(path, headers=auth_headers) + assert authed.status_code == 200 + assert authed.data == b'%PDF-1.4 invoice' + disposition = authed.headers.get('Content-Disposition', '') + assert 'attachment' in disposition + assert 'Dell invoice 4471.pdf' in disposition + + +def test_rejects_an_unsupported_type(client, db, auth_headers): + w = _warranty(db) + + response = client.post(f'/api/warranty/{w.warrantyid}/proof', + data={'file': (io.BytesIO(b'MZ'), 'setup.exe')}, + content_type='multipart/form-data', + headers=auth_headers) + + assert response.status_code == 400 + assert 'Unsupported document type' in response.get_data(as_text=True) + + +def test_rejects_an_oversize_document(client, db, auth_headers, monkeypatch): + from plugins.warranty.api import routes + monkeypatch.setattr(routes, 'MAX_PROOF_BYTES', 512) + w = _warranty(db) + + response = client.post(f'/api/warranty/{w.warrantyid}/proof', + data={'file': (io.BytesIO(b'x' * 2048), 'scan.pdf')}, + content_type='multipart/form-data', + headers=auth_headers) + + assert response.status_code == 400 + assert 'the limit is' in response.get_data(as_text=True) + + +def test_replacing_keeps_one_document_per_warranty(client, db, auth_headers): + import glob + import os + from flask import current_app + w = _warranty(db) + + client.post(f'/api/warranty/{w.warrantyid}/proof', + data={'file': (io.BytesIO(b'first'), 'scan.png')}, + content_type='multipart/form-data', headers=auth_headers) + client.post(f'/api/warranty/{w.warrantyid}/proof', + data={'file': (io.BytesIO(b'%PDF second'), 'invoice.pdf')}, + content_type='multipart/form-data', headers=auth_headers) + + proofdir = os.path.join(current_app.instance_path, 'warrantyproofs') + files = glob.glob(os.path.join(proofdir, f'warranty-{w.warrantyid}.*')) + assert len(files) == 1 and files[0].endswith('.pdf') + + +def test_removing_clears_both_columns(client, db, auth_headers): + w = _warranty(db) + client.post(f'/api/warranty/{w.warrantyid}/proof', + data={'file': (io.BytesIO(b'%PDF'), 'invoice.pdf')}, + content_type='multipart/form-data', headers=auth_headers) + + response = client.delete(f'/api/warranty/{w.warrantyid}/proof', + headers=auth_headers) + + body = response.get_json()['data'] + assert body['proofurl'] is None + assert body['prooffilename'] is None