Import-API hardening: network device IP endpoint + communicationtypes seeder

The legacy-import surface (docs/IMPORT-API.md) is the schema-agnostic contract
every adopting site targets; these close two gaps found while mapping the WJ
classic import.

Network device IP: POST/PUT /api/network now accept an `ipaddress` and
materialize a primary Communication (mirroring the printer route), and GET
(list + detail + create/update result) surface it. Previously a network
device's IP - which lives in the communications table, not on the extension -
had no HTTP import path at all.

communicationtypes seed: `flask seed reference-data` now seeds the eight
canonical communication types (IP/Serial/Network_Interface/USB/Parallel/VNC/
FTP/DNC), which IMPORT-API.md already documents as a prerequisite. The IP type
must exist before any asset import so printer/network routes can attach an IP.
There is no CRUD endpoint for these, so seeding is the only path.

Tests: network create/update IP upsert + GET surfacing + seed creates IP type.
204 targeted tests pass; naming + pyflakes green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

plugins/network/api/routes.py

@@ -3,7 +3,7 @@
 from flask import Blueprint, request
 from flask_jwt_extended import jwt_required
 
-from shopdb.api import db, Asset, AssetType, Vendor, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
+from shopdb.api import db, Asset, AssetType, Vendor, Communication, CommunicationType, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
 
 from ..models import NetworkDevice, NetworkDeviceType, Subnet, VLAN
 
@@ -12,6 +12,31 @@ from shopdb.api import require_permission, apply_import_timestamps
 network_bp = Blueprint('network', __name__)
 
 
+def _primary_ip(assetid):
+    """The asset's primary IP address string, or None. Mirrors the printer
+    route: a device's IP lives in a Communication row, not on the extension."""
+    comm = (Communication.query.filter_by(assetid=assetid, isprimary=True).first()
+            or Communication.query.filter_by(assetid=assetid).first())
+    return comm.ipaddress if comm else None
+
+
+def _upsert_primary_ip(asset, ip):
+    """Create/update/clear the asset's primary IP Communication from an
+    ipaddress string. No-op if the IP CommunicationType is not seeded."""
+    ip = (ip or '').strip()
+    if not ip:
+        return
+    comm = Communication.query.filter_by(assetid=asset.assetid, isprimary=True).first()
+    if comm:
+        comm.ipaddress = ip
+        return
+    ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
+    if ip_comtype:
+        db.session.add(Communication(
+            assetid=asset.assetid, comtypeid=ip_comtype.comtypeid,
+            ipaddress=ip, isprimary=True))
+
+
 # =============================================================================
 # Network Device Types
 # =============================================================================
@@ -233,6 +258,7 @@ def list_network_devices():
     for netdev in items:
         item = netdev.asset.to_dict() if netdev.asset else {}
         item['networkdevice'] = netdev.to_dict()
+        item['ipaddress'] = _primary_ip(netdev.assetid)
         data.append(item)
 
     return paginated_response(data, page, per_page, total)
@@ -253,6 +279,7 @@ def get_network_device(device_id: int):
 
     result = netdev.asset.to_dict() if netdev.asset else {}
     result['networkdevice'] = netdev.to_dict()
+    result['ipaddress'] = _primary_ip(netdev.assetid)
 
     return success_response(result)
 
@@ -272,6 +299,7 @@ def get_network_device_by_asset(asset_id: int):
 
     result = netdev.asset.to_dict() if netdev.asset else {}
     result['networkdevice'] = netdev.to_dict()
+    result['ipaddress'] = _primary_ip(netdev.assetid)
 
     return success_response(result)
 
@@ -291,6 +319,7 @@ def get_network_device_by_hostname(hostname: str):
 
     result = netdev.asset.to_dict() if netdev.asset else {}
     result['networkdevice'] = netdev.to_dict()
+    result['ipaddress'] = _primary_ip(netdev.assetid)
 
     return success_response(result)
 
@@ -380,6 +409,11 @@ def create_network_device():
     db.session.add(netdev)
     db.session.flush()
 
+    # Attach the primary IP as a Communication when supplied (a device's IP is
+    # not a NetworkDevice column). Mirrors the printer create route.
+    if data.get('ipaddress'):
+        _upsert_primary_ip(asset, data['ipaddress'])
+
     # Preserve legacy timestamps in import mode (no-op otherwise)
     apply_import_timestamps(asset, data)
 
@@ -391,6 +425,7 @@ def create_network_device():
 
     result = asset.to_dict()
     result['networkdevice'] = netdev.to_dict()
+    result['ipaddress'] = _primary_ip(asset.assetid)
 
     return success_response(result, message='Network device created', http_code=201)
 
@@ -465,11 +500,16 @@ def update_network_device(device_id: int):
         AuditLog.log('updated', 'NetworkDevice', entityid=netdev.networkdeviceid,
                      entityname=netdev.hostname or asset.assetnumber, changes=changes)
 
+    # Upsert the primary IP Communication when an ipaddress is supplied.
+    if 'ipaddress' in data:
+        _upsert_primary_ip(asset, data['ipaddress'])
+
     apply_import_timestamps(asset, data)
     db.session.commit()
 
     result = asset.to_dict()
     result['networkdevice'] = netdev.to_dict()
+    result['ipaddress'] = _primary_ip(asset.assetid)
 
     return success_response(result, message='Network device updated')
 

shopdb/cli/__init__.py

@@ -127,7 +127,8 @@ def seed_cli():
 def seed_reference_data():
     """Seed reference data (model types, statuses, etc.)."""
     from shopdb.extensions import db
-    from shopdb.core.models import ModelType, OperatingSystem, AssetStatus, LocationType
+    from shopdb.core.models import (ModelType, OperatingSystem, AssetStatus,
+                                     LocationType, CommunicationType)
     from shopdb.core.models.relationship import RelationshipType
 
     # Model types (type the vendor models catalog)
@@ -178,6 +179,23 @@ def seed_reference_data():
         if not LocationType.query.filter_by(locationtype=lt).first():
             db.session.add(LocationType(locationtype=lt, isactive=True))
 
+    # Communication types (how an asset is reached / its interfaces). The IP
+    # type is what the printer + network create routes attach an ipaddress to,
+    # so it must exist before any asset import.
+    comm_types = [
+        ('IP', 'IP address / network reachable'),
+        ('Serial', 'Serial (RS-232) connection'),
+        ('Network_Interface', 'Physical network interface (MAC/port)'),
+        ('USB', 'USB connection'),
+        ('Parallel', 'Parallel port connection'),
+        ('VNC', 'VNC remote access'),
+        ('FTP', 'FTP file transfer'),
+        ('DNC', 'Direct numerical control link'),
+    ]
+    for comtype, description in comm_types:
+        if not CommunicationType.query.filter_by(comtype=comtype).first():
+            db.session.add(CommunicationType(comtype=comtype, description=description))
+
     # Operating systems
     os_list = [
         {'osname': 'Windows 10', 'osversion': '10.0'},

tests/test_plugins/test_network_ipaddress.py

@@ -0,0 +1,68 @@
+"""Network device create/update attaches a primary IP Communication.
+
+Classic network devices carry their IP in the communications table, and the
+legacy import needs an HTTP path to set it. The network create/update routes
+accept an `ipaddress` and materialize a primary Communication (mirroring the
+printer route); GET surfaces it back.
+"""
+
+import pytest
+
+
+@pytest.fixture
+def network_setup(db):
+    from shopdb.core.models import AssetType, CommunicationType
+    db.session.add(AssetType(assettype='network_device', pluginname='network',
+                             tablename='networkdevices', description='Network devices'))
+    db.session.add(CommunicationType(comtype='IP', description='IP'))
+    db.session.commit()
+
+
+def test_create_network_device_with_ipaddress(client, db, auth_headers, network_setup):
+    """POST /api/network with an ipaddress creates a primary Communication and
+    echoes the IP back."""
+    from shopdb.core.models import Communication
+    resp = client.post('/api/network', json={
+        'assetnumber': 'SW-CORE-01', 'name': 'Core switch',
+        'hostname': 'sw-core-01', 'ipaddress': '10.129.22.101',
+    }, headers=auth_headers)
+    assert resp.status_code == 201, resp.get_json()
+    data = resp.get_json()['data']
+    assert data['ipaddress'] == '10.129.22.101'
+
+    assetid = data['assetid']
+    comm = Communication.query.filter_by(assetid=assetid, isprimary=True).first()
+    assert comm is not None and comm.ipaddress == '10.129.22.101'
+
+    # GET surfaces it too.
+    got = client.get(f"/api/network/{data['networkdevice']['networkdeviceid']}",
+                     headers=auth_headers)
+    assert got.get_json()['data']['ipaddress'] == '10.129.22.101'
+
+
+def test_update_network_device_ipaddress_upserts(client, db, auth_headers, network_setup):
+    """PUT changes the IP in place without creating a second Communication."""
+    from shopdb.core.models import Communication
+    created = client.post('/api/network', json={
+        'assetnumber': 'AP-01', 'name': 'AP', 'ipaddress': '10.0.0.5',
+    }, headers=auth_headers).get_json()['data']
+    devid = created['networkdevice']['networkdeviceid']
+
+    updated = client.put(f'/api/network/{devid}', json={'ipaddress': '10.0.0.9'},
+                         headers=auth_headers)
+    assert updated.status_code == 200, updated.get_json()
+    assert updated.get_json()['data']['ipaddress'] == '10.0.0.9'
+
+    comms = Communication.query.filter_by(assetid=created['assetid']).all()
+    assert len(comms) == 1 and comms[0].ipaddress == '10.0.0.9'
+
+
+def test_seed_reference_data_creates_ip_commtype(app, db):
+    """The reference-data seed now creates the IP communication type the asset
+    import depends on."""
+    from shopdb.core.models import CommunicationType
+    runner = app.test_cli_runner()
+    assert runner.invoke(args=['seed', 'reference-data']).exit_code in (0, None)
+    with app.app_context():
+        assert CommunicationType.query.filter_by(comtype='IP').first() is not None
+        assert CommunicationType.query.count() >= 8