Attach proof of cover to a warranty
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 stayed in somebody's mailbox until they left. Two columns rather than one: the served URL of the stored document, and the name the vendor sent it under, because "Dell invoice 4471.pdf" is what a person recognises a year later and "warranty-12.pdf" is not. The download route sends the original name back. Authenticated in both directions, unlike an asset photo: an invoice carries pricing and a service tag. One document per warranty, replacing any prior extension so a re-upload as .pdf does not leave the old .png behind claiming to be current. Capped at 25MB - a certificate is a document, not a disk image. Office formats are allowed because purchase records genuinely arrive as .msg and .xlsx, not only as PDFs.
This commit is contained in:
cproudlock
@@ -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-<id><ext>, one per warranty, with the vendor's own
|
||||
# filename kept alongside so a person recognises it later.
|
||||
# =============================================================================
|
||||
|
||||
@warranty_bp.route('/<int:warrantyid>/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/<path:filename>', 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('/<int:warrantyid>/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')
|
||||
|
||||
@@ -151,6 +151,21 @@
|
||||
<label>Notes</label>
|
||||
<textarea v-model="form.notes" class="form-control" rows="2"></textarea>
|
||||
</div>
|
||||
|
||||
<div class="form-group">
|
||||
<label>Proof of cover</label>
|
||||
<div v-if="proofName" class="proof-current">
|
||||
<a v-if="proofUrl" :href="proofUrl" @click.prevent="downloadProof">{{ proofName }}</a>
|
||||
<span v-else>{{ proofName }}</span>
|
||||
<button type="button" class="btn btn-sm btn-secondary" @click="removeProof"
|
||||
:disabled="busyProof">Remove</button>
|
||||
</div>
|
||||
<input type="file" class="form-control" :accept="PROOF_ACCEPT" @change="onProofPicked" />
|
||||
<small class="muted">
|
||||
Invoice, certificate or a saved email - up to 25MB. Downloading
|
||||
needs a login, since it carries pricing and a service tag.
|
||||
</small>
|
||||
</div>
|
||||
<div v-if="error" class="error-message">{{ error }}</div>
|
||||
</div>
|
||||
<div class="modal-footer">
|
||||
@@ -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; }
|
||||
|
||||
50
plugins/warranty/migrations/versions/0002_proof_document.py
Normal file
50
plugins/warranty/migrations/versions/0002_proof_document.py
Normal file
@@ -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)
|
||||
@@ -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']),
|
||||
|
||||
110
tests/test_plugins/test_warranty_proof.py
Normal file
110
tests/test_plugins/test_warranty_proof.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user