diff --git a/plugins/network/api/routes.py b/plugins/network/api/routes.py index 66f2c0a..06657cb 100644 --- a/plugins/network/api/routes.py +++ b/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') diff --git a/shopdb/cli/__init__.py b/shopdb/cli/__init__.py index 7d2c0d9..3d99133 100644 --- a/shopdb/cli/__init__.py +++ b/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'}, diff --git a/tests/test_plugins/test_network_ipaddress.py b/tests/test_plugins/test_network_ipaddress.py new file mode 100644 index 0000000..1fd9fe8 --- /dev/null +++ b/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