ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dashboarddefaults: key display mappings by stable FQDN (from BIOS serial), IP fallback

Browse Source 
A display's DHCP IP can change; its FQDN (F<serial>.<domain>, domain from the
display_fqdn_domain setting) is stable and the collector already reports the
serial. Add a nullable unique fqdn column (varchar191 so the index fits utf8mb4
without innodb_large_prefix), make ipaddress nullable, and require fqdn OR ip.
visitor-location + display-role resolve by FQDN first, then IP; create/update
accept fqdn. Core migration 7d31, verified up/down/idempotent on MySQL 5.6.
'Business unit' wording -> 'location' in the validation messages.
This commit is contained in:
cproudlock
2026-07-29 07:20:52 -04:00
parent b539e36096
commit 3ba808028c
4 changed files with 160 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
migrations/versions/7d31_dashboarddefault_fqdn.py Normal file
View File

@@ -0,0 +1,62 @@
"""dashboarddefaults: add fqdn, make ipaddress nullable.
Map a display PC to its role/location by its STABLE FQDN (derived from the BIOS
serial, F<serial>.<domain>) instead of only its DHCP-changeable IP. IP stays as
a nullable fallback. Idempotent.
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '7d31_dashboarddefault_fqdn'
down_revision = '7d30_apitoken_resourcescopes'
branch_labels = None
depends_on = None
def _has_column(bind, table, column):
inspector = sa.inspect(bind)
if table not in inspector.get_table_names():
return False
return column in {c['name'] for c in inspector.get_columns(table)}
def _col_nullable(bind, table, column):
inspector = sa.inspect(bind)
for c in inspector.get_columns(table):
if c['name'] == column:
return c.get('nullable', True)
return True
def _has_index(bind, table, index):
inspector = sa.inspect(bind)
if table not in inspector.get_table_names():
return False
return index in {i['name'] for i in inspector.get_indexes(table)}
def upgrade():
bind = op.get_bind()
if not _has_column(bind, 'dashboarddefaults', 'fqdn'):
op.add_column('dashboarddefaults',
sa.Column('fqdn', sa.String(191), nullable=True))
if not _has_index(bind, 'dashboarddefaults', 'idx_dashboarddefaults_fqdn'):
op.create_index('idx_dashboarddefaults_fqdn', 'dashboarddefaults',
['fqdn'], unique=True)
# IP is now optional (an FQDN-only mapping has no IP).
if not _col_nullable(bind, 'dashboarddefaults', 'ipaddress'):
op.alter_column('dashboarddefaults', 'ipaddress',
existing_type=sa.String(50), nullable=True)
def downgrade():
bind = op.get_bind()
if _has_index(bind, 'dashboarddefaults', 'idx_dashboarddefaults_fqdn'):
op.drop_index('idx_dashboarddefaults_fqdn',
table_name='dashboarddefaults')
if _has_column(bind, 'dashboarddefaults', 'fqdn'):
op.drop_column('dashboarddefaults', 'fqdn')
# ipaddress is left nullable: reverting to NOT NULL could fail on FQDN-only
# rows that legitimately have no IP.

61
shopdb/core/api/dashboarddefaults.py
View File

@@ -32,6 +32,18 @@ def _serialize(default):
return data
def _resolve_default(fqdn, ipaddress):
"""Find the active mapping for a display: FQDN first (stable), then IP."""
default = None
if fqdn:
default = DashboardDefault.query.filter_by(
fqdn=fqdn, isactive=True).first()
if not default and ipaddress:
default = DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first()
return default
@dashboarddefaults_bp.route('/visitor-location', methods=['GET'])
def visitor_location():
"""Resolve the business unit for the calling display by its IP.
@@ -39,16 +51,18 @@ def visitor_location():
Unauthenticated: kiosks/lobby displays hit this. Returns the mapped
business unit, or a null businessunitid when the IP is not mapped.
"""
fqdn = (request.args.get('fqdn') or '').strip().lower() or None
ipaddress = request.args.get('ipaddress') or _request_ip()
default = DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first()
default = _resolve_default(fqdn, ipaddress)
if not default:
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'businessunitid': None,
'businessunit': None,
})
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'businessunitid': default.businessunitid,
'businessunit': default.businessunit.businessunit
@@ -64,11 +78,12 @@ def display_role():
(dashboard / lobby / partskiosk), the frontend path it maps to, and the
business unit for the dashboard role. Unmapped IP -> null role.
"""
fqdn = (request.args.get('fqdn') or '').strip().lower() or None
ipaddress = request.args.get('ipaddress') or _request_ip()
default = DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first()
default = _resolve_default(fqdn, ipaddress)
if not default:
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'role': None,
'path': None,
@@ -76,6 +91,7 @@ def display_role():
'businessunit': None,
})
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'role': default.displayrole,
'path': default.displaypath,
@@ -100,36 +116,42 @@ def list_defaults():
def create_default():
"""Create a visitor-IP -> business-unit mapping."""
data = request.get_json() or {}
ipaddress = (data.get('ipaddress') or '').strip()
fqdn = (data.get('fqdn') or '').strip().lower() or None
ipaddress = (data.get('ipaddress') or '').strip() or None
role = (data.get('displayrole') or 'dashboard').strip()
businessunitid = data.get('businessunitid')
if not ipaddress:
return error_response(ErrorCodes.VALIDATION_ERROR, 'ipaddress is required')
if not fqdn and not ipaddress:
return error_response(ErrorCodes.VALIDATION_ERROR,
'fqdn or ipaddress is required')
if role not in DISPLAY_ROLES:
return error_response(ErrorCodes.VALIDATION_ERROR,
f'displayrole must be one of {", ".join(DISPLAY_ROLES)}')
# Only the dashboard role needs a business unit.
# Only the dashboard role needs a business unit (location).
if role == 'dashboard':
if not businessunitid:
return error_response(ErrorCodes.VALIDATION_ERROR,
'businessunitid is required for the dashboard role')
'a location is required for the dashboard role')
if not db.session.get(BusinessUnit, businessunitid):
return error_response(ErrorCodes.NOT_FOUND, 'Business unit not found',
return error_response(ErrorCodes.NOT_FOUND, 'Location not found',
http_code=404)
else:
businessunitid = None
if DashboardDefault.query.filter_by(ipaddress=ipaddress, isactive=True).first():
if fqdn and DashboardDefault.query.filter_by(fqdn=fqdn, isactive=True).first():
return error_response(ErrorCodes.CONFLICT,
f"{fqdn} is already mapped", http_code=409)
if ipaddress and DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first():
return error_response(ErrorCodes.CONFLICT,
f"IP {ipaddress} is already mapped", http_code=409)
default = DashboardDefault(ipaddress=ipaddress, displayrole=role,
default = DashboardDefault(fqdn=fqdn, ipaddress=ipaddress, displayrole=role,
businessunitid=businessunitid,
description=data.get('description'))
db.session.add(default)
db.session.flush()
AuditLog.log('created', 'DashboardDefault', entityid=default.dashboarddefaultid,
entityname=ipaddress)
entityname=(fqdn or ipaddress))
db.session.commit()
return success_response(_serialize(default), message='Mapping created',
http_code=201)
@@ -156,17 +178,22 @@ def update_default(default_id):
return error_response(ErrorCodes.NOT_FOUND, 'Business unit not found',
http_code=404)
default.businessunitid = data['businessunitid']
if 'ipaddress' in data and data['ipaddress']:
default.ipaddress = data['ipaddress'].strip()
if 'fqdn' in data:
default.fqdn = (data.get('fqdn') or '').strip().lower() or None
if 'ipaddress' in data:
default.ipaddress = (data.get('ipaddress') or '').strip() or None
if 'description' in data:
default.description = data['description']
# Non-dashboard roles carry no business unit; dashboard needs one.
if not default.fqdn and not default.ipaddress:
return error_response(ErrorCodes.VALIDATION_ERROR,
'fqdn or ipaddress is required')
# Non-dashboard roles carry no location; dashboard needs one.
if default.displayrole != 'dashboard':
default.businessunitid = None
elif not default.businessunitid:
return error_response(ErrorCodes.VALIDATION_ERROR,
'businessunitid is required for the dashboard role')
'a location is required for the dashboard role')
db.session.commit()
return success_response(_serialize(default), message='Mapping updated')

35
shopdb/core/models/dashboarddefault.py
View File

@@ -18,14 +18,41 @@ DISPLAY_ROLE_PATHS = {
}
DISPLAY_ROLES = tuple(DISPLAY_ROLE_PATHS.keys())
# GE device naming: a PC's DNS name is 'F' + its BIOS serial under the device
# domain, e.g. FABC1234.device.geaerospace.net. The domain is a setting so other
# sites can point elsewhere; the collector already reports the serial, so the
# server derives the stable FQDN without the kiosk having to report it.
DEFAULT_DISPLAY_FQDN_DOMAIN = 'device.geaerospace.net'
def derive_display_fqdn(serialnumber):
"""FQDN for a display PC from its BIOS serial: F<serial>.<domain> (lower).
Returns None when there is no serial. Domain from the display_fqdn_domain
setting, falling back to DEFAULT_DISPLAY_FQDN_DOMAIN.
"""
serial = (serialnumber or '').strip()
if not serial:
return None
from .setting import Setting
domain = (Setting.get('display_fqdn_domain', DEFAULT_DISPLAY_FQDN_DOMAIN)
or DEFAULT_DISPLAY_FQDN_DOMAIN).strip().strip('.')
return f'F{serial}.{domain}'.lower()
class DashboardDefault(BaseModel):
"""Maps a display PC IP address to its display role (+ business unit)."""
"""Maps a display PC (by FQDN, or IP) to its display role (+ location)."""
__tablename__ = 'dashboarddefaults'
dashboarddefaultid = db.Column(db.Integer, primary_key=True)
ipaddress = db.Column(db.String(50), unique=True, nullable=False)
# Which display the IP drives. Only the dashboard role uses businessunitid.
# A mapping is keyed by FQDN (stable, DHCP-proof) and/or IP. At least one is
# required (enforced in the API). FQDN is preferred; IP is the fallback for a
# manual entry or a PC not yet reporting a serial.
# 191 = the utf8mb4-safe unique-index length (191*4 < 767) so the index does
# not depend on innodb_large_prefix; FQDNs (F<serial>.<domain>) fit easily.
fqdn = db.Column(db.String(191), unique=True, nullable=True, index=True)
ipaddress = db.Column(db.String(50), unique=True, nullable=True)
# Which display the mapping drives. Only the dashboard role uses businessunitid.
displayrole = db.Column(db.String(20), nullable=False, default='dashboard')
businessunitid = db.Column(
db.Integer,
@@ -41,4 +68,4 @@ class DashboardDefault(BaseModel):
return DISPLAY_ROLE_PATHS.get(self.displayrole)
def __repr__(self):
return f"<DashboardDefault {self.ipaddress} -> {self.displayrole}>"
return f"<DashboardDefault {self.fqdn or self.ipaddress} -> {self.displayrole}>"

24
tests/test_core/test_dashboarddefaults.py
View File

@@ -129,7 +129,29 @@ def test_dashboard_role_requires_businessunit(client, db, auth_headers):
'ipaddress': '10.20.30.55', 'displayrole': 'dashboard',
}, headers=auth_headers)
assert resp.status_code == 400
assert 'businessunitid' in resp.get_json()['data']['error']['message'].lower()
assert 'location' in resp.get_json()['data']['error']['message'].lower()
def test_fqdn_mapping_create_and_resolve(client, db, auth_headers):
# A mapping can be keyed by FQDN (no IP), and /display-role resolves it.
resp = client.post('/api/dashboarddefaults', json={
'fqdn': 'FABC123.device.geaerospace.net', 'displayrole': 'lobby',
}, headers=auth_headers)
assert resp.status_code == 201, resp.data
assert resp.get_json()['data']['fqdn'] == 'fabc123.device.geaerospace.net'
# resolve by FQDN (case-insensitive)
r = client.get('/api/dashboarddefaults/display-role?fqdn=fabc123.device.geaerospace.net')
body = r.get_json()['data']
assert body['role'] == 'lobby'
assert body['path'] == '/tv'
def test_mapping_requires_fqdn_or_ip(client, db, auth_headers):
resp = client.post('/api/dashboarddefaults', json={'displayrole': 'lobby'},
headers=auth_headers)
assert resp.status_code == 400
assert 'fqdn or ipaddress' in resp.get_json()['data']['error']['message'].lower()
def test_invalid_role_rejected(client, db, auth_headers):