Refuse a value the database cannot store, instead of quietly changing it

Wave 2 of the remediation plan. Eight findings from the feature test run were
the same defect twice: a write that reports success while storing something
other than what was sent.

The first half is a date parser that swallowed its own error. Two plugins
carried an identical private _parse_date returning None on ValueError, so an
unparseable date and a deliberately cleared one produced the same write.
PUT /api/warranty/<id> with enddate '30-06-2027' answered 200 "Warranty
updated" and stored NULL, dropping the record out of every expiry surface on
the strength of a typo; the measuring-tool calibration date did the same and
flipped the tool to 'unknown', out of the calibration report and the map
overlay. shopdb/utils/datevalues.py now holds one strict parser on the contract
surface: an explicitly empty value clears the field, anything that is not a
date is refused with a 400 naming the field, and those are different intents
with different answers. The vendor-lookup paths keep a lenient wrapper on
purpose - a provider changing its date shape must not fail the refresh of an
otherwise good record, and no person typed that value.

The second half is why the truncation findings were possible at all. MySQL was
running non-strict: sql_mode on the 5.6 server is NO_ENGINE_SUBSTITUTION with
no STRICT_TRANS_TABLES, so it accepted writes it could not store and adjusted
them. A 257-character username became its first 100 characters (and the next
distinct one then collided with it); an over-long asset number was cut at 50; a
malformed releasedate became '0000-00-00' and permanently broke every read of
that application. All with a 201. The mode is now set on the CONNECTION rather
than left to the server, so a site whose my.cnf disagrees still gets it.

That trades silent corruption for a DataError, which uncaught is a 500 - and a
500 says the server is broken when the truth is that the value does not fit. An
error handler maps it to a 400 naming the column where MySQL names one.
Verified by hand against MySQL 5.6: POST /api/users with a 257-character
username now answers 400 "That value is too long for username." and stores
nothing, where it previously stored a truncated row and answered 201.

Twelve tests. The date guard is verified by planting: restore the swallowing
parser and four of them fail. The DataError tests raise the error directly
because the suite runs on SQLite, which does not enforce VARCHAR length at all -
there is no value that would trigger it there, which is precisely why this went
unnoticed.

Scope: product
This commit is contained in:
cproudlock
2026-09-03 13:58:41 -04:00
parent aa8028f96e
commit 12392a361e
9 changed files with 419 additions and 34 deletions

View File

@@ -20,6 +20,7 @@ from shopdb.api import (
success_response, error_response, paginated_response, ErrorCodes, success_response, error_response, paginated_response, ErrorCodes,
get_pagination_params, paginate_query, get_pagination_params, paginate_query,
require_permission, apply_import_timestamps, require_permission, apply_import_timestamps,
DateValueError, parse_date_fields,
) )
from ..models import MeasuringTool, MeasuringToolType, derive_status, STATUS_COLORS from ..models import MeasuringTool, MeasuringToolType, derive_status, STATUS_COLORS
@@ -27,16 +28,6 @@ from ..models import MeasuringTool, MeasuringToolType, derive_status, STATUS_COL
measuringtools_bp = Blueprint('measuringtools', __name__) measuringtools_bp = Blueprint('measuringtools', __name__)
def _parse_date(value):
"""Accept 'YYYY-MM-DD' (or None/empty) -> date or None."""
if not value:
return None
try:
return datetime.strptime(str(value)[:10], '%Y-%m-%d').date()
except (ValueError, TypeError):
return None
def _merged(tool, today=None): def _merged(tool, today=None):
"""Asset core dict with the measuringtools extension nested under 'measuringtool'.""" """Asset core dict with the measuringtools extension nested under 'measuringtool'."""
result = tool.asset.to_dict() if tool.asset else {} result = tool.asset.to_dict() if tool.asset else {}
@@ -240,6 +231,12 @@ def _asset_type_id():
def create_tool(): def create_tool():
"""Create a measuring tool: one Asset core row + one extension row.""" """Create a measuring tool: one Asset core row + one extension row."""
data = request.get_json() or {} data = request.get_json() or {}
# Refused, not nulled - see the note at the update site.
try:
dates = parse_date_fields(
data, ('lastcalibrationdate', 'nextcalibrationdate'))
except DateValueError as bad:
return error_response(ErrorCodes.VALIDATION_ERROR, str(bad))
if not data.get('assetnumber'): if not data.get('assetnumber'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'assetnumber is required') return error_response(ErrorCodes.VALIDATION_ERROR, 'assetnumber is required')
if Asset.query.filter_by(assetnumber=data['assetnumber']).first(): if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
@@ -275,8 +272,8 @@ def create_tool():
assetid=asset.assetid, assetid=asset.assetid,
measuringtooltypeid=data.get('measuringtooltypeid'), measuringtooltypeid=data.get('measuringtooltypeid'),
calibrationintervaldays=data.get('calibrationintervaldays'), calibrationintervaldays=data.get('calibrationintervaldays'),
lastcalibrationdate=_parse_date(data.get('lastcalibrationdate')), lastcalibrationdate=dates.get('lastcalibrationdate'),
nextcalibrationdate=_parse_date(data.get('nextcalibrationdate')), nextcalibrationdate=dates.get('nextcalibrationdate'),
calibrationprovider=(data.get('calibrationprovider') or '').strip() or None, calibrationprovider=(data.get('calibrationprovider') or '').strip() or None,
notes=(data.get('notes') or '').strip() or None, notes=(data.get('notes') or '').strip() or None,
) )
@@ -309,6 +306,12 @@ def update_tool(tool_id: int):
return error_response(ErrorCodes.NOT_FOUND, return error_response(ErrorCodes.NOT_FOUND,
f'Measuring tool {tool_id} not found', http_code=404) f'Measuring tool {tool_id} not found', http_code=404)
data = request.get_json() or {} data = request.get_json() or {}
# Refused, not nulled - see the note at the update site.
try:
dates = parse_date_fields(
data, ('lastcalibrationdate', 'nextcalibrationdate'))
except DateValueError as bad:
return error_response(ErrorCodes.VALIDATION_ERROR, str(bad))
asset = tool.asset asset = tool.asset
# An UPDATE may not blank the asset number. Create validates it and the # An UPDATE may not blank the asset number. Create validates it and the
@@ -339,9 +342,11 @@ def update_tool(tool_id: int):
if getattr(tool, key) != data[key]: if getattr(tool, key) != data[key]:
changes[key] = {'old': getattr(tool, key), 'new': data[key]} changes[key] = {'old': getattr(tool, key), 'new': data[key]}
setattr(tool, key, data[key]) setattr(tool, key, data[key])
for key in ('lastcalibrationdate', 'nextcalibrationdate'): # Refused, not nulled: a mistyped calibration date used to store NULL and
if key in data: # answer 200, which flipped the tool to 'unknown' and dropped it out of the
setattr(tool, key, _parse_date(data[key])) # calibration report and the map overlay.
for key, parsed in dates.items():
setattr(tool, key, parsed)
if changes: if changes:
AuditLog.log('updated', 'MeasuringTool', entityid=tool.measuringtoolid, AuditLog.log('updated', 'MeasuringTool', entityid=tool.measuringtoolid,

View File

@@ -19,6 +19,7 @@ from shopdb.api import (
db, Asset, db, Asset,
success_response, error_response, ErrorCodes, success_response, error_response, ErrorCodes,
require_permission, require_permission,
DateValueError, parse_date, parse_date_fields,
) )
from ..models import Warranty, WarrantyAsset from ..models import Warranty, WarrantyAsset
@@ -42,13 +43,16 @@ def _proof_dir():
return os.path.join(current_app.instance_path, 'warrantyproofs') return os.path.join(current_app.instance_path, 'warrantyproofs')
def _parse_date(value): def _vendor_date(value):
"""Accept 'YYYY-MM-DD' (or None/empty) -> date or None.""" """A date off a VENDOR response, which this code does not control.
if not value:
return None Kept lenient on purpose, unlike a date a person typed: a provider that
starts returning a shape we do not parse must not fail the refresh of an
otherwise good record. The field is simply left as it was.
"""
try: try:
return datetime.strptime(value[:10], '%Y-%m-%d').date() return parse_date(value)
except (ValueError, TypeError): except DateValueError:
return None return None
@@ -203,13 +207,20 @@ def create_warranty():
vendor = (data.get('vendor') or '').strip() vendor = (data.get('vendor') or '').strip()
if not vendor: if not vendor:
return error_response(ErrorCodes.VALIDATION_ERROR, 'vendor is required') return error_response(ErrorCodes.VALIDATION_ERROR, 'vendor is required')
# Refused, not nulled. A mistyped end date used to store NULL and answer
# 200, which dropped the warranty out of every expiry surface on the
# strength of a typo.
try:
dates = parse_date_fields(data, ('startdate', 'enddate'))
except DateValueError as bad:
return error_response(ErrorCodes.VALIDATION_ERROR, str(bad))
warranty = Warranty( warranty = Warranty(
vendor=vendor, vendor=vendor,
servicetag=(data.get('servicetag') or '').strip() or None, servicetag=(data.get('servicetag') or '').strip() or None,
provider=(data.get('provider') or 'manual').strip().lower(), provider=(data.get('provider') or 'manual').strip().lower(),
servicelevel=(data.get('servicelevel') or '').strip() or None, servicelevel=(data.get('servicelevel') or '').strip() or None,
startdate=_parse_date(data.get('startdate')), startdate=dates.get('startdate'),
enddate=_parse_date(data.get('enddate')), enddate=dates.get('enddate'),
notes=(data.get('notes') or '').strip() or None, notes=(data.get('notes') or '').strip() or None,
) )
_apply_links(warranty, data.get('assetids')) _apply_links(warranty, data.get('assetids'))
@@ -225,6 +236,10 @@ def update_warranty(warrantyid):
if not warranty: if not warranty:
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404) return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
data = request.get_json() or {} data = request.get_json() or {}
try:
dates = parse_date_fields(data, ('startdate', 'enddate'))
except DateValueError as bad:
return error_response(ErrorCodes.VALIDATION_ERROR, str(bad))
if 'vendor' in data: if 'vendor' in data:
warranty.vendor = (data['vendor'] or '').strip() or warranty.vendor warranty.vendor = (data['vendor'] or '').strip() or warranty.vendor
if 'servicetag' in data: if 'servicetag' in data:
@@ -233,10 +248,8 @@ def update_warranty(warrantyid):
warranty.provider = (data['provider'] or 'manual').strip().lower() warranty.provider = (data['provider'] or 'manual').strip().lower()
if 'servicelevel' in data: if 'servicelevel' in data:
warranty.servicelevel = (data['servicelevel'] or '').strip() or None warranty.servicelevel = (data['servicelevel'] or '').strip() or None
if 'startdate' in data: for field, parsed in dates.items():
warranty.startdate = _parse_date(data['startdate']) setattr(warranty, field, parsed)
if 'enddate' in data:
warranty.enddate = _parse_date(data['enddate'])
if 'notes' in data: if 'notes' in data:
warranty.notes = (data['notes'] or '').strip() or None warranty.notes = (data['notes'] or '').strip() or None
if 'isactive' in data: if 'isactive' in data:
@@ -279,9 +292,9 @@ def refresh_warranty(warrantyid):
if result.get('servicelevel'): if result.get('servicelevel'):
warranty.servicelevel = result['servicelevel'] warranty.servicelevel = result['servicelevel']
if result.get('startdate'): if result.get('startdate'):
warranty.startdate = _parse_date(result['startdate']) warranty.startdate = _vendor_date(result['startdate'])
if result.get('enddate'): if result.get('enddate'):
warranty.enddate = _parse_date(result['enddate']) warranty.enddate = _vendor_date(result['enddate'])
warranty.lastcheckeddate = datetime.now(timezone.utc).replace(tzinfo=None) warranty.lastcheckeddate = datetime.now(timezone.utc).replace(tzinfo=None)
db.session.commit() db.session.commit()
return success_response(_warranty_payload(warranty), message='Warranty refreshed') return success_response(_warranty_payload(warranty), message='Warranty refreshed')
@@ -363,8 +376,8 @@ def sync_dell():
existing.provider = 'dell' existing.provider = 'dell'
existing.vendor = existing.vendor or 'Dell' existing.vendor = existing.vendor or 'Dell'
existing.servicelevel = found.get('servicelevel') existing.servicelevel = found.get('servicelevel')
existing.startdate = _parse_date(found.get('startdate')) existing.startdate = _vendor_date(found.get('startdate'))
existing.enddate = _parse_date(found.get('enddate')) existing.enddate = _vendor_date(found.get('enddate'))
existing.servicetag = tag existing.servicetag = tag
existing.lastcheckeddate = now existing.lastcheckeddate = now
updated += 1 updated += 1
@@ -372,8 +385,8 @@ def sync_dell():
warranty = Warranty( warranty = Warranty(
vendor='Dell', provider='dell', servicetag=tag, vendor='Dell', provider='dell', servicetag=tag,
servicelevel=found.get('servicelevel'), servicelevel=found.get('servicelevel'),
startdate=_parse_date(found.get('startdate')), startdate=_vendor_date(found.get('startdate')),
enddate=_parse_date(found.get('enddate')), enddate=_vendor_date(found.get('enddate')),
lastcheckeddate=now, lastcheckeddate=now,
) )
warranty.links.append(WarrantyAsset(assetid=assetid)) warranty.links.append(WarrantyAsset(assetid=assetid))

View File

@@ -368,6 +368,38 @@ def register_error_handlers(app: Flask):
return error_response(ErrorCodes.VALIDATION_ERROR, return error_response(ErrorCodes.VALIDATION_ERROR,
'The database rejected that change.', http_code=400) 'The database rejected that change.', http_code=400)
# STRICT_TRANS_TABLES (see config.SQLALCHEMY_ENGINE_OPTIONS) turns what
# MySQL used to silently adjust into an error: a value too long for its
# column, a date that is not a date, a number that is not a number. That is
# the behaviour we want - the alternative is a 201 for a row the caller did
# not send - but the error is the CALLER's, so it must not read as a server
# fault. Without this it surfaced as a 500 with a traceback in the log.
from sqlalchemy.exc import DataError
@app.errorhandler(DataError)
def handle_data_error(error):
from .extensions import db
db.session.rollback()
message = str(getattr(error, 'orig', error))
app.logger.warning('data error: %s', message)
column = _column_from_data_error(message)
if 'too long' in message.lower():
return error_response(
ErrorCodes.VALIDATION_ERROR,
'That value is too long for %s.' % (column or 'this field'),
http_code=400)
if 'incorrect date' in message.lower() or \
'incorrect datetime' in message.lower():
return error_response(
ErrorCodes.VALIDATION_ERROR,
'That is not a valid date for %s.' % (column or 'this field'),
http_code=400)
return error_response(
ErrorCodes.VALIDATION_ERROR,
'The database rejected that value%s.'
% (' for %s' % column if column else ''),
http_code=400)
@app.errorhandler(ShopDBException) @app.errorhandler(ShopDBException)
def handle_shopdb_exception(error): def handle_shopdb_exception(error):
http_codes = { http_codes = {
@@ -410,6 +442,20 @@ def register_error_handlers(app: Flask):
) )
def _column_from_data_error(message):
"""Pull the column name out of a MySQL data error, when it names one.
MySQL says "Data too long for column 'username' at row 1"; SQLite says
nothing of the sort. Best effort on purpose - a message that names the
field is much more useful than a generic one, and a message that names
nothing is still better than a 500.
"""
import re
found = re.search(r"column '([^']+)'", message or '')
return found.group(1) if found else None
def register_frontend_routes(app: Flask): def register_frontend_routes(app: Flask):
"""Serve Vue frontend static files.""" """Serve Vue frontend static files."""
frontend_dist = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'frontend', 'dist') frontend_dist = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'frontend', 'dist')

View File

@@ -110,6 +110,14 @@ from shopdb.utils.uploads import send_upload, harden_upload_response
# nothing. See shopdb/utils/clientip.py. # nothing. See shopdb/utils/clientip.py.
from shopdb.utils.clientip import client_ip, strip_port from shopdb.utils.clientip import client_ip, strip_port
# Strict date parsing. On the contract surface because every plugin that stores
# a date needs it and the two that rolled their own both swallowed the error.
from shopdb.utils.datevalues import (
DateValueError,
parse_date,
parse_date_fields,
)
# Import-mode helpers: preserve legacy timestamps during a bulk data import # Import-mode helpers: preserve legacy timestamps during a bulk data import
from shopdb.utils.import_mode import ( from shopdb.utils.import_mode import (
apply_import_timestamps, apply_import_timestamps,
@@ -350,6 +358,9 @@ __all__ = [
# Response + pagination helpers # Response + pagination helpers
'success_response', 'success_response',
'error_response', 'error_response',
'DateValueError',
'parse_date',
'parse_date_fields',
'paginated_response', 'paginated_response',
'ErrorCodes', 'ErrorCodes',
'get_pagination_params', 'get_pagination_params',

View File

@@ -38,7 +38,26 @@ class Config:
# Pin the client connection to utf8mb4 so reads/writes match the # Pin the client connection to utf8mb4 so reads/writes match the
# utf8mb4 schema regardless of the server default charset. TestingConfig # utf8mb4 schema regardless of the server default charset. TestingConfig
# (SQLite) replaces this whole dict, so the charset arg never reaches it. # (SQLite) replaces this whole dict, so the charset arg never reaches it.
'connect_args': {'charset': 'utf8mb4'}, #
# STRICT_TRANS_TABLES is the other half, and it is a correctness
# setting rather than a preference. Without it MySQL accepts a write it
# cannot store and adjusts it: a 257-character username becomes the
# first 100 characters, a malformed date becomes '0000-00-00', an
# over-long asset number is cut at 50 and then collides with a
# different one. Every one of those was found in the product answering
# 200 or 201, because nothing in the stack knew the value had changed.
#
# Set on the CONNECTION rather than left to the server, so a site whose
# my.cnf says otherwise still gets it - the sql_mode seen on the 5.6
# dev server was NO_ENGINE_SUBSTITUTION alone. NO_ENGINE_SUBSTITUTION is
# kept alongside it, since dropping it would change how a CREATE TABLE
# naming an unavailable engine behaves.
'connect_args': {
'charset': 'utf8mb4',
'init_command':
"SET SESSION sql_mode='STRICT_TRANS_TABLES,"
"NO_ENGINE_SUBSTITUTION'",
},
} }
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-secret-key-change-in-production') JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-secret-key-change-in-production')

View File

@@ -0,0 +1,82 @@
"""Parsing a date out of a request body, strictly.
WHY THIS EXISTS
Two plugins carried an identical private `_parse_date` that swallowed the
error:
try:
return datetime.strptime(str(value)[:10], '%Y-%m-%d').date()
except (ValueError, TypeError):
return None
So an unparseable date and a deliberately cleared one produced the same
result, and the write went ahead. `PUT /api/warranty/<id>` with enddate
'not-a-date' answered 200 "Warranty updated" and stored NULL - the warranty
then vanished from every expiry surface, silently, on the strength of a typo.
The measuring-tool calibration date did the same thing and dropped the tool out
of the calibration report.
The rule here: an EXPLICIT empty value clears the field, and anything that is
not a date is refused. Those are different intents and they get different
answers. Callers turn `DateValueError` into a 400 naming the field, so the
person who mistyped it finds out at the point of the mistake rather than at the
next audit.
Accepted forms are deliberately narrow - 'YYYY-MM-DD', and the ISO-8601
datetime forms whose leading ten characters are a date, since that is what the
frontend's date inputs and every vendor API in the tree emit. Anything looser
is a guess about what somebody meant.
"""
from datetime import date, datetime
class DateValueError(ValueError):
"""A supplied value was neither empty nor a date this accepts.
Carries the field name so the error message can say which one, which is the
whole point of refusing rather than nulling.
"""
def __init__(self, field, value):
self.field = field
self.value = value
super().__init__(
'%s must be a date as YYYY-MM-DD (got %r)' % (field, value))
def parse_date(value, field='date'):
"""Return a date, or None for an explicitly empty value.
Raises DateValueError for anything else. None and '' mean "clear it";
they are the only values that produce None.
"""
if value is None:
return None
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
if not isinstance(value, str):
# A number or a list is not a mistyped date, it is the wrong type.
raise DateValueError(field, value)
text = value.strip()
if not text:
return None
try:
return datetime.strptime(text[:10], '%Y-%m-%d').date()
except ValueError:
raise DateValueError(field, value)
def parse_date_fields(data, fields):
"""Parse several date fields out of a payload at once.
Only keys PRESENT in the payload are returned, so a caller can tell "not
mentioned" (leave it alone) from "sent as empty" (clear it) - a distinction
a plain `data.get(...)` throws away, and the reason the alias-clearing bug
existed on the PC form.
"""
return {field: parse_date(data[field], field)
for field in fields if field in data}

View File

@@ -0,0 +1,99 @@
"""A write the database refuses is the CALLER's error, not a server fault.
The engine now runs with STRICT_TRANS_TABLES (see
shopdb/config.py: SQLALCHEMY_ENGINE_OPTIONS). Before that, MySQL silently
adjusted a write it could not store: a 257-character username became the first
100 characters, an unparseable date became '0000-00-00' and then broke every
read of that row, an over-long asset number was cut at 50 and collided with a
different one. All of them answered 200 or 201, because nothing in the stack
knew the value had changed.
Strict mode turns those into errors, which is the point - but an uncaught
DataError is a 500, and a 500 says "the server is broken" when the truth is
"that value does not fit". This pins the mapping.
These tests raise the error directly rather than through a column, because the
suite runs on SQLite, which does not enforce VARCHAR length at all: there is no
value that would trigger it here. The behaviour against the real engine was
checked by hand on MySQL 5.6 - `POST /api/users` with a 257-character username
returns 400 "That value is too long for username." and stores nothing, where it
previously stored a truncated row.
"""
import pytest
from sqlalchemy.exc import DataError
from shopdb import _column_from_data_error
# A private app, because routes can only be mounted before the first request
# and the session-wide one has already served plenty.
@pytest.fixture(scope='module')
def raising_app():
from shopdb import create_app
application = create_app('testing')
application.add_url_rule(
'/__test__/raise', 'raise_data_error',
lambda: (_ for _ in ()).throw(
_data_error(application.config['TEST_DATA_ERROR_MESSAGE'])))
return application
def _raise(app, message):
"""Answer a request whose route raises DataError, so the app's own handler
is what produces the response."""
app.config['TEST_DATA_ERROR_MESSAGE'] = message
return app.test_client().get('/__test__/raise')
def _data_error(message):
return DataError('INSERT INTO x', {}, Exception(message))
def test_a_value_too_long_is_a_400_naming_the_column(raising_app):
response = _raise(
raising_app,
"(1406, \"Data too long for column 'username' at row 1\")")
assert response.status_code == 400
body = response.get_json()
assert body['data']['error']['code'] == 'VALIDATION_ERROR'
assert 'too long' in body['data']['error']['message']
assert 'username' in body['data']['error']['message']
def test_a_bad_date_is_a_400_naming_the_column(raising_app):
response = _raise(
raising_app,
"(1292, \"Incorrect date value: 'not-a-date' for column "
"'releasedate' at row 1\")")
assert response.status_code == 400
assert 'releasedate' in response.get_json()['data']['error']['message']
def test_an_unrecognised_data_error_is_still_a_400_not_a_500(raising_app):
"""The point is the CLASS of answer. A message shape nobody anticipated
must not fall back to 'internal server error'."""
response = _raise(raising_app, '(9999, "something new")')
assert response.status_code == 400
assert response.get_json()['data']['error']['code'] == 'VALIDATION_ERROR'
@pytest.mark.parametrize('message,expected', [
("Data too long for column 'assetnumber' at row 1", 'assetnumber'),
("Incorrect date value: 'x' for column 'enddate' at row 1", 'enddate'),
('no column mentioned at all', None),
('', None),
(None, None),
])
def test_the_column_name_is_pulled_out_when_the_engine_names_one(message,
expected):
assert _column_from_data_error(message) == expected
def test_the_strict_sql_mode_is_actually_set_on_the_mysql_connection():
"""The whole class of fix rests on this one connect_arg, and it is exactly
the kind of line a later edit drops without noticing."""
from shopdb.config import Config
connect_args = Config.SQLALCHEMY_ENGINE_OPTIONS['connect_args']
assert 'STRICT_TRANS_TABLES' in connect_args['init_command']

View File

@@ -488,3 +488,68 @@ def test_asset_presentation_route_token():
entries = MeasuringToolsPlugin().get_asset_presentation() entries = MeasuringToolsPlugin().get_asset_presentation()
entry = next(e for e in entries if e['assettype'] == 'measuring_tool') entry = next(e for e in entries if e['assettype'] == 'measuring_tool')
assert entry['route'] == '/measuringtools/by-asset/{assetid}' assert entry['route'] == '/measuringtools/by-asset/{assetid}'
# --- a date that is not a date -----------------------------------------------
#
# _parse_date used to swallow the ValueError and return None, so a typo in the
# calibration date answered 200 "Measuring tool updated", stored NULL, flipped
# the tool to 'unknown', and dropped it out of the calibration report and the
# map overlay. Clearing a date and mistyping one are different intents and now
# get different answers.
def test_a_mistyped_calibration_date_is_refused_not_stored_as_null(
client, auth_headers):
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-DATE-1',
'name': 'Date guard caliper',
'statusid': _status_id(client),
'nextcalibrationdate': '2027-01-01',
})
assert created.status_code == 201, created.get_json()
toolid = created.get_json()['data']['measuringtool']['measuringtoolid']
refused = client.put(f'/api/measuringtools/{toolid}', headers=auth_headers,
json={'nextcalibrationdate': 'not-a-date'})
assert refused.status_code == 400, refused.get_json()
assert 'nextcalibrationdate' in \
refused.get_json()['data']['error']['message']
# The stored date is untouched - the whole point of refusing.
after = client.get(f'/api/measuringtools/{toolid}',
headers=auth_headers).get_json()['data']
assert after['measuringtool']['nextcalibrationdate'] == '2027-01-01'
assert after['measuringtool']['calibrationstatus'] != 'unknown'
def test_an_explicitly_empty_calibration_date_still_clears_it(client,
auth_headers):
"""Clearing is a real intent and must keep working, or the fix above just
trades one broken behaviour for another."""
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-DATE-2',
'name': 'Clearable caliper',
'statusid': _status_id(client),
'nextcalibrationdate': '2027-01-01',
})
toolid = created.get_json()['data']['measuringtool']['measuringtoolid']
cleared = client.put(f'/api/measuringtools/{toolid}', headers=auth_headers,
json={'nextcalibrationdate': ''})
assert cleared.status_code == 200, cleared.get_json()
after = client.get(f'/api/measuringtools/{toolid}',
headers=auth_headers).get_json()['data']
assert after['measuringtool']['nextcalibrationdate'] is None
def test_a_mistyped_date_is_refused_at_create_too(client, auth_headers):
refused = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-DATE-3',
'name': 'Never created',
'statusid': _status_id(client),
'lastcalibrationdate': '01/02/2026',
})
assert refused.status_code == 400, refused.get_json()
listed = client.get('/api/measuringtools?search=MT-DATE-3',
headers=auth_headers).get_json()['data']
assert listed == [], 'nothing may be created by a refused request'

View File

@@ -108,3 +108,48 @@ def test_removing_clears_both_columns(client, db, auth_headers):
body = response.get_json()['data'] body = response.get_json()['data']
assert body['proofurl'] is None assert body['proofurl'] is None
assert body['prooffilename'] is None assert body['prooffilename'] is None
# --- a date that is not a date -----------------------------------------------
#
# PUT /api/warranty/<id> with an unparseable enddate answered 200 "Warranty
# updated" and stored NULL, which drops the record out of every expiry surface
# it feeds. The local _parse_date swallowed the ValueError, so a typo and a
# deliberate clear produced the same write.
def _warranty_via_api(client, auth_headers, **fields):
payload = {'vendor': 'Acme', 'servicetag': 'WD-1', 'enddate': '2027-06-30'}
payload.update(fields)
created = client.post('/api/warranty', headers=auth_headers, json=payload)
assert created.status_code == 201, created.get_json()
return created.get_json()['data']['warrantyid']
def test_a_mistyped_end_date_is_refused_not_stored_as_null(client, db,
auth_headers):
warrantyid = _warranty_via_api(client, auth_headers)
refused = client.put(f'/api/warranty/{warrantyid}', headers=auth_headers,
json={'enddate': '30-06-2027'})
assert refused.status_code == 400, refused.get_json()
assert 'enddate' in refused.get_json()['data']['error']['message']
kept = db.session.get(Warranty, warrantyid)
assert kept.enddate is not None, 'the stored cover date must survive a typo'
assert str(kept.enddate) == '2027-06-30'
def test_an_explicitly_empty_end_date_still_clears_it(client, db, auth_headers):
warrantyid = _warranty_via_api(client, auth_headers, servicetag='WD-2')
cleared = client.put(f'/api/warranty/{warrantyid}', headers=auth_headers,
json={'enddate': ''})
assert cleared.status_code == 200, cleared.get_json()
assert db.session.get(Warranty, warrantyid).enddate is None
def test_a_mistyped_date_is_refused_at_create_too(client, db, auth_headers):
refused = client.post('/api/warranty', headers=auth_headers,
json={'vendor': 'Acme', 'servicetag': 'WD-3',
'startdate': 'yesterday'})
assert refused.status_code == 400, refused.get_json()
assert Warranty.query.filter_by(servicetag='WD-3').first() is None