diff --git a/pyproject.toml b/pyproject.toml index 9c7f786..744cbab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] -name = "bacnet-scan-tool" -version = "0.1.0rc0" +name = "bacnet-scan-api" +version = "0.1.0rc2" description = "" authors = [ {name = "C. Allwardt",email = "3979063+craig8@users.noreply.github.com"} @@ -11,7 +11,7 @@ requires-python = ">=3.10,<4.0" [tool.poetry] packages = [ - {include="bacnet_scan_tool", from="src"}] + {include="bacnet_scan_api", from="src"}] [build-system] requires = ["poetry-core>=2.0.0,<3.0.0"] @@ -24,12 +24,8 @@ pydantic = ">=2.10.6,<3.0.0" sqlmodel = ">=0.0.8,<1.0.0" click = ">=8.1.8,<9.0.0" uvicorn = ">=0.34.0,<0.35.0" -#protocol-proxy = { git = "https://github.com/riley206-pnnl/lib-protocol-proxy.git", rev = "bacnet-async-upgrade" } -#protocol-proxy-bacnet = { git = "https://github.com/riley206-pnnl/lib-protocol-proxy-bacnet.git", rev = "main" } -#protocol-proxy-bacnet = { path = "../lib-protocol-proxy-bacnet" } -#protocol-proxy = { path = "../lib-protocol-proxy" } -protocol-proxy-bacnet = ">=v2.0.0rc2" -protocol-proxy = ">=v2.0.0rc1" +protocol-proxy = ">=2.0.0rc2" +protocol-proxy-bacnet = ">=2.0.0rc2" bacpypes3 = "^0.0.102" psutil = "^7.0.0" gevent = "^25.5.1" @@ -37,3 +33,10 @@ python-multipart = "^0.0.20" netifaces = "^0.11.0" bac0 = "^2025.6.10" rdflib = "^6.0" + +[dependency-groups] +dev = [ + "pytest (>=8.4.2,<9.0.0)", + "pytest-asyncio (>=1.2.0,<2.0.0)", + "httpx (>=0.28.1,<0.29.0)" +] diff --git a/src/bacnet_scan_tool/__init__.py b/src/bacnet_scan_api/__init__.py similarity index 100% rename from src/bacnet_scan_tool/__init__.py rename to src/bacnet_scan_api/__init__.py diff --git a/src/bacnet_scan_tool/main.py b/src/bacnet_scan_api/main.py similarity index 97% rename from src/bacnet_scan_tool/main.py rename to src/bacnet_scan_api/main.py index a619afd..d617359 100644 --- a/src/bacnet_scan_tool/main.py +++ b/src/bacnet_scan_api/main.py @@ -31,18 +31,31 @@ async def start_proxy(local_device_address: Optional[str] = Form(None)): Returns status and address. """ try: + print(f"[start_proxy] Received local_device_address: {local_device_address}") if not local_device_address: try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) local_device_address = s.getsockname()[0] s.close() - except Exception: + print(f"[start_proxy] Auto-detected IP: {local_device_address}") + except Exception as e: + print(f"[start_proxy] Auto-detection failed: {e}") return ProxyResponse( status="error", - error= - "Could not auto-detect local IP address. Please specify manually." + error=f"Could not auto-detect local IP address. Please specify manually. Error: {str(e)}" ) + + # Validate we have a valid IP address + if not local_device_address: + print("[start_proxy] No IP address after detection") + return ProxyResponse( + status="error", + error="No IP address provided or detected." + ) + + print(f"[start_proxy] Using IP address: {local_device_address}") + if hasattr(app.state, "bacnet_manager") and app.state.bacnet_manager: await app.state.bacnet_manager.stop() if hasattr(app.state, @@ -58,14 +71,13 @@ async def start_proxy(local_device_address: Optional[str] = Form(None)): app.state.bacnet_manager.inbound_server.serve_forever()) app.state.bacnet_proxy_peer = await app.state.bacnet_manager.get_proxy( - (local_device_address, 0), + (local_device_address, 0), local_device_address=local_device_address) app.state.bacnet_proxy_local_address = local_device_address - asyncio.create_task( - app.state.bacnet_manager.wait_peer_registered( - peer=app.state.bacnet_proxy_peer, timeout=5)) - await asyncio.sleep(1) + await app.state.bacnet_manager.wait_peer_registered( + peer=app.state.bacnet_proxy_peer, timeout=5) + return ProxyResponse(status="done", address=local_device_address) except Exception as e: return ProxyResponse(status="error", error=str(e)) diff --git a/src/bacnet_scan_tool/models.py b/src/bacnet_scan_api/models.py similarity index 100% rename from src/bacnet_scan_tool/models.py rename to src/bacnet_scan_api/models.py diff --git a/src/bacnet_scan_tool/network_discover_ping.py b/src/bacnet_scan_api/network_discover_ping.py similarity index 100% rename from src/bacnet_scan_tool/network_discover_ping.py rename to src/bacnet_scan_api/network_discover_ping.py diff --git a/tests/backend/test_endpoints.py b/tests/backend/test_endpoints.py index d3cedfd..c89f838 100644 --- a/tests/backend/test_endpoints.py +++ b/tests/backend/test_endpoints.py @@ -1,179 +1,626 @@ +""" +Pytest unit tests for BACnet Scan Tool FastAPI endpoints +""" import pytest +import json +import subprocess +from unittest.mock import Mock, AsyncMock, patch, MagicMock from fastapi.testclient import TestClient +import asyncio -# Import your FastAPI app -from bacnet_scan_tool.main import app +from bacnet_scan_api.main import app +from bacnet_scan_api.models import ( + ProxyResponse, IPAddress, ScanResponse, PropertyReadResponse, + DevicePropertiesResponse, WhoIsResponse, PingResponse, + ObjectListNamesResponse, SavedScansResponse, ScannedPointsResponse +) -# Create test client -client = TestClient(app) -# API endpoint constants -TOOL_NAME = "bacnet" -SCAN_ENDPOINT = f"/{TOOL_NAME}/scan/start" -DEVICES_ENDPOINT = "/devices" -POINTS_ENDPOINT = "/points" -VALUES_ENDPOINT = "/points/values" -META_ENDPOINT = "/points/meta" -TAGS_ENDPOINT = "/tags" -WRITE_ENDPOINT = "/write" +@pytest.fixture +def client(): + """Create a test client for the FastAPI app""" + return TestClient(app) + + +@pytest.fixture +def mock_bacnet_manager(): + """Mock BACnet manager with all necessary attributes""" + manager = AsyncMock() + manager.start = AsyncMock() + manager.stop = AsyncMock() + manager.get_proxy = AsyncMock() + manager.send = AsyncMock() + manager.wait_peer_registered = AsyncMock() + manager.inbound_server = MagicMock() + manager.inbound_server.serve_forever = AsyncMock() + return manager + + +@pytest.fixture +def mock_bacnet_peer(): + """Mock BACnet peer""" + peer = MagicMock() + peer.address = ("192.168.1.173", 47808) + return peer + + +class TestStartProxy: + """Tests for /start_proxy endpoint""" + + def test_start_proxy_with_address(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test starting proxy with explicit local address""" + with patch('bacnet_scan_tool.main.AsyncioProtocolProxyManager') as MockManager: + MockManager.get_manager.return_value = mock_bacnet_manager + mock_bacnet_manager.get_proxy.return_value = mock_bacnet_peer + + response = client.post("/start_proxy", data={"local_device_address": "192.168.1.173/24"}) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + assert data["address"] == "192.168.1.173/24" + + def test_start_proxy_auto_detect(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test starting proxy with auto-detected address""" + with patch('bacnet_scan_tool.main.AsyncioProtocolProxyManager') as MockManager, \ + patch('bacnet_scan_tool.main.discover_networks_for_bacnet') as mock_discover: + + MockManager.get_manager.return_value = mock_bacnet_manager + mock_bacnet_manager.get_proxy.return_value = mock_bacnet_peer + mock_discover.return_value = {"interface_networks": ["192.168.1.0/24"]} + + response = client.post("/start_proxy", data={}) + + assert response.status_code == 200 + data = response.json() + assert data["status"] in ["done", "error"] + + def test_start_proxy_error_handling(self, client): + """Test error handling when proxy fails to start""" + with patch('bacnet_scan_tool.main.AsyncioProtocolProxyManager') as MockManager: + MockManager.get_manager.side_effect = Exception("Connection failed") + + response = client.post("/start_proxy", data={"local_device_address": "192.168.1.173/24"}) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + assert "Connection failed" in data["error"] + + +class TestStopProxy: + """Tests for /stop_proxy endpoint""" + + def test_stop_proxy_success(self, client, mock_bacnet_manager): + """Test successfully stopping the proxy""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_server_task = AsyncMock() + app.state.bacnet_proxy_peer = MagicMock() + app.state.bacnet_proxy_local_address = "192.168.1.173/24" + + response = client.post("/stop_proxy") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + + def test_stop_proxy_when_not_running(self, client): + """Test stopping proxy when nothing is running""" + # Clear any existing state + if hasattr(app.state, 'bacnet_manager'): + delattr(app.state, 'bacnet_manager') + + response = client.post("/stop_proxy") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + + +class TestGetHostIP: + """Tests for /get_host_ip endpoint""" + + def test_get_host_ip_linux(self, client): + """Test getting host IP on Linux system""" + with patch('subprocess.check_output') as mock_subprocess: + # Simulate Linux environment + mock_subprocess.side_effect = [ + FileNotFoundError, # No ipconfig.exe (not WSL) + b"default via 192.168.1.1 dev eth0", # ip route + ] + + response = client.get("/get_host_ip") + + # Should succeed or fail gracefully + assert response.status_code in [200, 500] + + def test_get_host_ip_with_hostname(self, client): + """Test getting host IP using hostname -I""" + with patch('subprocess.check_output') as mock_subprocess: + mock_subprocess.side_effect = [ + FileNotFoundError, # No ipconfig.exe + subprocess.CalledProcessError(1, 'cmd'), # ip route fails + b"192.168.1.100 172.17.0.1", # hostname -I + ] + + response = client.get("/get_host_ip") + + if response.status_code == 200: + data = response.json() + assert "address" in data + + +class TestScanSubnet: + """Tests for /bacnet/scan_subnet endpoint""" + + def test_scan_subnet_success(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test successful subnet scan""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + # Mock scan response + mock_devices = [ + { + "pduSource": "192.168.1.248", + "deviceIdentifier": ["device", 3056211], + "maxAPDULengthAccepted": 1024, + "segmentationSupported": "segmented-both", + "vendorID": 842 + } + ] + mock_bacnet_manager.send.return_value = json.dumps(mock_devices).encode('utf8') + + response = client.post("/bacnet/scan_subnet", data={"subnet": "192.168.1.0/24"}) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + assert "devices" in data + assert data["ips_scanned"] == 254 + + def test_scan_subnet_with_all_parameters(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test scan with all optional parameters""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_bacnet_manager.send.return_value = json.dumps([]).encode('utf8') + + response = client.post("/bacnet/scan_subnet", data={ + "subnet": "192.168.1.0/24", + "whois_timeout": 5.0, + "port": 47808, + "low_id": 0, + "high_id": 1000, + "enable_brute_force": True, + "semaphore_limit": 20, + "max_duration": 300.0 + }) + + assert response.status_code == 200 + + def test_scan_subnet_no_proxy(self, client): + """Test scan when proxy is not registered""" + app.state.bacnet_proxy_peer = None + + response = client.post("/bacnet/scan_subnet", data={"subnet": "192.168.1.0/24"}) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + assert "Proxy not registered" in data["error"] + + def test_scan_subnet_timeout(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test scan timeout handling""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + # Mock timeout response + timeout_response = {"status": "error", "error": "Operation timed out after 280s"} + mock_bacnet_manager.send.return_value = json.dumps(timeout_response).encode('utf8') + + response = client.post("/bacnet/scan_subnet", data={"subnet": "192.168.1.0/24"}) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + assert "timed out" in data["error"] + + +class TestReadProperty: + """Tests for /read_property endpoint""" + + def test_read_property_success(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test successful property read""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + # Mock property value + mock_value = {"value": 72.5, "type": "Real"} + mock_bacnet_manager.send.return_value = json.dumps(mock_value).encode('utf8') + + response = client.post("/read_property", data={ + "device_address": "192.168.1.248", + "object_identifier": "analog-input,1", + "property_identifier": "present-value" + }) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + assert "result" in data # read_property returns result, not value directly + + def test_read_property_with_array_index(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test property read with array index""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_bacnet_manager.send.return_value = json.dumps({"value": "Test"}).encode('utf8') + + response = client.post("/read_property", data={ + "device_address": "192.168.1.248", + "object_identifier": "device,3056211", + "property_identifier": "object-list", + "property_array_index": 1 + }) + + assert response.status_code == 200 + + def test_read_property_no_proxy(self, client): + """Test read when proxy is not registered""" + app.state.bacnet_proxy_peer = None + + response = client.post("/read_property", data={ + "device_address": "192.168.1.248", + "object_identifier": "analog-input,1", + "property_identifier": "present-value" + }) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + -# Test data constants -TEST_IP_RANGE = "192.168.1.0/24" +class TestWriteProperty: + """Tests for /write_property endpoint""" + + def test_write_property_success(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test successful property write""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_bacnet_manager.send.return_value = json.dumps({"status": "done"}).encode('utf8') + + # write_property endpoint expects query parameters, not JSON body + response = client.post("/write_property", params={ + "device_address": "192.168.1.248", + "object_identifier": "analog-value,1", + "property_identifier": "present-value", + "value": 75.0, + "priority": 16 + }) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + + def test_write_property_with_array_index(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test write with array index""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_bacnet_manager.send.return_value = json.dumps({"status": "done"}).encode('utf8') + + # write_property endpoint expects query parameters, not JSON body + response = client.post("/write_property", params={ + "device_address": "192.168.1.248", + "object_identifier": "analog-value,1", + "property_identifier": "present-value", + "value": 75.0, + "priority": 16, + "property_array_index": 1 + }) + + assert response.status_code == 200 +class TestWhoIs: + """Tests for /bacnet/who_is endpoint""" + + def test_who_is_success(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test successful Who-Is request""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_devices = [ + { + "pduSource": "192.168.1.248", + "deviceIdentifier": ["device", 3056211], + "maxAPDULengthAccepted": 1024, + "segmentationSupported": "segmented-both", + "vendorID": 842 + } + ] + mock_bacnet_manager.send.return_value = json.dumps(mock_devices).encode('utf8') + + response = client.post("/bacnet/who_is", data={ + "device_instance_low": 0, + "device_instance_high": 4194303, + "dest": "192.168.1.255" + }) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + assert "devices" in data + + def test_who_is_dict_response(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test Who-Is with dict response format""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_response = {"devices": [{"pduSource": "192.168.1.248"}]} + mock_bacnet_manager.send.return_value = json.dumps(mock_response).encode('utf8') + + response = client.post("/bacnet/who_is", data={ + "device_instance_low": 0, + "device_instance_high": 4194303, + "dest": "192.168.1.255" + }) + + assert response.status_code == 200 + + +class TestReadDeviceAll: + """Tests for /bacnet/read_device_all endpoint""" + + def test_read_device_all_success(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test reading all device properties""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_properties = { + "object-name": "Test Device", + "description": "Test Description", + "model-name": "Test Model", + "vendor-id": 842 + } + mock_bacnet_manager.send.return_value = json.dumps(mock_properties).encode('utf8') + + response = client.post("/bacnet/read_device_all", data={ + "device_address": "192.168.1.248", + "device_object_identifier": "device,3056211" + }) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + assert "properties" in data + + +class TestReadObjectListNames: + """Tests for /bacnet/read_object_list_names endpoint""" + + def test_read_object_list_names_success(self, client, mock_bacnet_manager, mock_bacnet_peer): + """Test reading object list with names""" + app.state.bacnet_manager = mock_bacnet_manager + app.state.bacnet_proxy_peer = mock_bacnet_peer + + mock_response = { + "status": "done", + "object_list_names": { + "analog-input,1": { + "object-name": "Temperature Sensor", + "units": "degreesCelsius" + } + }, + "pagination": { + "current_page": 1, + "page_size": 100, + "total_objects": 1, + "total_pages": 1 + } + } + mock_bacnet_manager.send.return_value = json.dumps(mock_response).encode('utf8') + + response = client.post("/bacnet/read_object_list_names", data={ + "device_address": "192.168.1.248", + "device_object_identifier": "device,3056211", + "page": 1, + "page_size": 100 + }) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + + def test_read_object_list_names_pagination_invalid(self, client): + """Test invalid pagination parameters""" + response = client.post("/bacnet/read_object_list_names", data={ + "device_address": "192.168.1.248", + "device_object_identifier": "device,3056211", + "page": 0, + "page_size": 100 + }) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + + +class TestPingIP: + """Tests for /ping_ip endpoint""" + + @pytest.mark.asyncio + async def test_ping_success(self, client): + """Test successful ping""" + with patch('asyncio.create_subprocess_exec') as mock_subprocess: + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"Reply from 192.168.1.1", b"") + mock_proc.returncode = 0 + mock_subprocess.return_value = mock_proc + + response = client.post("/ping_ip", data={"ip_address": "192.168.1.1"}) + + assert response.status_code == 200 + data = response.json() + assert "ip_address" in data + + def test_ping_failure(self, client): + """Test failed ping""" + with patch('asyncio.create_subprocess_exec') as mock_subprocess: + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"", b"Request timed out") + mock_proc.returncode = 1 + mock_subprocess.return_value = mock_proc + + response = client.post("/ping_ip", data={"ip_address": "192.168.1.1"}) + + assert response.status_code == 200 + + +class TestDiscoverNetworks: + """Tests for /discover_networks endpoint""" + + @pytest.mark.asyncio + async def test_discover_networks_success(self, client): + """Test successful network discovery""" + with patch('bacnet_scan_tool.main.discover_networks_for_bacnet') as mock_discover: + mock_discover.return_value = { + "interface_networks": ["192.168.1.0/24"], + "route_networks": ["10.0.0.0/8"] + } + + response = client.get("/discover_networks") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "done" + assert "networks" in data + + def test_discover_networks_with_verbose(self, client): + """Test network discovery with verbose output""" + with patch('bacnet_scan_tool.main.discover_networks_for_bacnet') as mock_discover: + mock_discover.return_value = {"interface_networks": ["192.168.1.0/24"]} + + response = client.get("/discover_networks?verbose=true") + + assert response.status_code == 200 + + +class TestCustomNetworks: + """Tests for custom network management endpoints""" + + def test_add_custom_network(self, client): + """Test adding a custom network""" + with patch('pathlib.Path.exists', return_value=False), \ + patch('builtins.open', create=True) as mock_open: + + response = client.post("/networks/add", data={"network": "192.168.2.0/24"}) + + assert response.status_code == 200 + data = response.json() + assert data["status"] in ["done", "error"] + + def test_add_invalid_network(self, client): + """Test adding invalid network format""" + response = client.post("/networks/add", data={"network": "invalid"}) + + assert response.status_code == 200 + data = response.json() + # Should handle invalid format gracefully + + def test_remove_custom_network(self, client): + """Test removing a custom network""" + mock_data = { + "custom_networks": ["192.168.2.0/24", "10.0.0.0/8"], + "added_dates": {"192.168.2.0/24": "2025-11-03T10:00:00", "10.0.0.0/8": "2025-11-03T11:00:00"} + } + + with patch('pathlib.Path.exists', return_value=True), \ + patch('builtins.open', create=True) as mock_open: + + # Mock file reading and writing + mock_file = MagicMock() + mock_file.__enter__.return_value.read.return_value = json.dumps(mock_data) + mock_open.return_value = mock_file + + # Use request method for DELETE with form data + response = client.request("DELETE", "/networks/remove", data={"network": "192.168.2.0/24"}) + + assert response.status_code == 200 + + def test_get_custom_networks(self, client): + """Test retrieving custom networks""" + with patch('pathlib.Path.exists', return_value=False): + response = client.get("/networks/custom") + + assert response.status_code == 200 + data = response.json() + assert "networks" in data + + +class TestRetrieveSavedScans: + """Tests for /retrieve_saved_scans endpoint""" + + def test_retrieve_saved_scans_success(self, client): + """Test retrieving saved scans""" + with patch('pathlib.Path.exists', return_value=True), \ + patch('builtins.open', create=True) as mock_open: + + mock_open.return_value.__enter__.return_value.read.return_value = '[]' + + response = client.get("/retrieve_saved_scans") + + assert response.status_code == 200 + data = response.json() + assert "devices" in data + + def test_retrieve_saved_scans_no_file(self, client): + """Test when no saved scans exist""" + with patch('pathlib.Path.exists', return_value=False): + response = client.get("/retrieve_saved_scans") + + assert response.status_code == 200 + data = response.json() + assert data["total_count"] == 0 + + +class TestRetrieveScannedPoints: + """Tests for /retrieve_scanned_points endpoint""" + + def test_retrieve_all_points(self, client): + """Test retrieving all scanned points""" + with patch('pathlib.Path.exists', return_value=True), \ + patch('builtins.open', create=True) as mock_open: + + mock_open.return_value.__enter__.return_value.read.return_value = '[]' + + response = client.get("/retrieve_scanned_points") + + assert response.status_code == 200 + data = response.json() + assert "points" in data + + def test_retrieve_points_by_device(self, client): + """Test retrieving points for specific device""" + with patch('pathlib.Path.exists', return_value=True), \ + patch('builtins.open', create=True) as mock_open: + + mock_open.return_value.__enter__.return_value.read.return_value = '[]' + + response = client.get("/retrieve_scanned_points?device_address=192.168.1.248") + + assert response.status_code == 200 + + +# Fixtures for async testing @pytest.fixture -def run_fake_scan(): - """Run a fake scan to populate data structures with test data""" - response = client.get(SCAN_ENDPOINT, - params={ - "ip_address": TEST_IP_RANGE, - "fake": "true" - }) - assert response.status_code == 200 - assert "device_id" in response.json() - return response.json()["device_id"] - - -def test_start_bacnet_discovery(): - """Test starting a BACnet scan""" - response = client.get(SCAN_ENDPOINT, - params={ - "ip_address": TEST_IP_RANGE, - "fake": "true" - }) - assert response.status_code == 200 - result = response.json() - assert "device_id" in result - assert "message" in result - - -def test_get_devices(run_fake_scan): - """Test retrieving devices after a scan""" - response = client.get(DEVICES_ENDPOINT) - assert response.status_code == 200 - result = response.json() - assert "devices" in result - assert len(result["devices"]) > 0 - # Check for expected device fields - device = result["devices"][0] - assert "id" in device - assert "ip_address" in device - - -def test_get_device_points(run_fake_scan): - """Test retrieving points for a specific device""" - device_id = run_fake_scan - response = client.get(f"{DEVICES_ENDPOINT}/{device_id}{POINTS_ENDPOINT}") - assert response.status_code == 200 - result = response.json() - assert "points" in result - assert len(result["points"]) > 0 - # Check for expected point fields - point = result["points"][0] - assert "id" in point - assert "name" in point - assert "device_id" in point - - -def test_get_point_values(run_fake_scan): - """Test retrieving values for all points of a device""" - device_id = run_fake_scan - response = client.get(f"{DEVICES_ENDPOINT}/{device_id}{VALUES_ENDPOINT}") - assert response.status_code == 200 - result = response.json() - assert "values" in result - # Values should be a dictionary with point_id keys - assert len(result["values"]) > 0 - # Get first point_id and value - point_id = list(result["values"].keys())[0] - point_data = result["values"][point_id] - assert "value" in point_data - - -def test_get_point_metadata(run_fake_scan): - """Test retrieving metadata for all points of a device""" - device_id = run_fake_scan - response = client.get(f"{DEVICES_ENDPOINT}/{device_id}{META_ENDPOINT}") - assert response.status_code == 200 - result = response.json() - assert "metadata" in result - # Metadata should be a dictionary with point_id keys - assert len(result["metadata"]) > 0 - - -def test_create_point_tag(run_fake_scan): - """Test creating a tag for a specific point""" - device_id = run_fake_scan - # Get a point ID first - points_response = client.get( - f"{DEVICES_ENDPOINT}/{device_id}{POINTS_ENDPOINT}") - assert points_response.status_code == 200 - point_id = points_response.json()["points"][0]["id"] - - # Create a tag - tag_data = {"name": "test_tag"} - response = client.post( - f"{DEVICES_ENDPOINT}/{device_id}/points/{point_id}{TAGS_ENDPOINT}", - json=tag_data) - assert response.status_code == 200 - result = response.json() - assert "tag_id" in result - assert "message" in result - - -def test_get_point_tags(run_fake_scan): - """Test retrieving tags for a specific point""" - device_id = run_fake_scan - # Get a point ID first - points_response = client.get( - f"{DEVICES_ENDPOINT}/{device_id}{POINTS_ENDPOINT}") - assert points_response.status_code == 200 - point_id = points_response.json()["points"][0]["id"] - - # First add a tag - tag_data = {"name": "test_tag"} - client.post( - f"{DEVICES_ENDPOINT}/{device_id}/points/{point_id}{TAGS_ENDPOINT}", - json=tag_data) - - # Get the tags - response = client.get( - f"{DEVICES_ENDPOINT}/{device_id}/points/{point_id}{TAGS_ENDPOINT}") - assert response.status_code == 200 - result = response.json() - assert "tags" in result - assert "test_tag" in result["tags"] - - -def test_write_point_value(run_fake_scan): - """Test writing a value to a specific point""" - device_id = run_fake_scan - # Get a point ID first - points_response = client.get( - f"{DEVICES_ENDPOINT}/{device_id}{POINTS_ENDPOINT}") - assert points_response.status_code == 200 - point_id = points_response.json()["points"][0]["id"] - - # Write a value to the point - write_data = {"value": 72.5} - response = client.put( - f"{DEVICES_ENDPOINT}/{device_id}/points/{point_id}{WRITE_ENDPOINT}", - json=write_data) - assert response.status_code == 200 - result = response.json() - assert "message" in result - - -def test_nonexistent_device(): - """Test accessing a device that doesn't exist""" - response = client.get(f"{DEVICES_ENDPOINT}/99999{POINTS_ENDPOINT}") - assert response.status_code == 404 - assert "not found" in response.json()["detail"].lower() - - -def test_invalid_tool_name(): - """Test using an invalid tool name""" - response = client.get("/invalid-tool/scan/start", - params={"ip_address": TEST_IP_RANGE}) - assert response.status_code == 400 - assert "unsupported tool" in response.json()["detail"].lower() +def event_loop(): + """Create an instance of the default event loop for the test session.""" + loop = asyncio.get_event_loop_policy().new_event_loop() + yield loop + loop.close() + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])