from unittest.mock import Mock, patch from django.test import TestCase, override_settings from django.utils import timezone from accounts.models import User from accounts.services.jwt_service import issue_jwt from vpn.models import Device, VpnLog from vpn.services.netmaker_client import NetmakerApiError, NetmakerClient, NetmakerConfigError class NetmakerClientTests(TestCase): @override_settings( NETMAKER_BASE_URL="https://netmaker.example.com", NETMAKER_API_TOKEN="secret", NETMAKER_NETWORK_ID="office", NETMAKER_ENROLLMENT_TAGS=["homevpn", "desktop", "vpn"], NETMAKER_ENROLLMENT_TTL_SECONDS=600, ) @patch("vpn.services.netmaker_client.uuid4") @patch("vpn.services.netmaker_client.requests.post") def test_create_one_time_join_token(self, post, uuid4): uuid4.return_value.hex = "abcdef1234567890" response = Mock() response.status_code = 200 response.json.return_value = {"token": "join-token"} post.return_value = response token = NetmakerClient().create_one_time_join_token("pc-01") self.assertEqual(token, "join-token") url = post.call_args.args[0] payload = post.call_args.kwargs["json"] headers = post.call_args.kwargs["headers"] self.assertEqual(url, "https://netmaker.example.com/api/v1/enrollment-keys") self.assertEqual(payload["name"], "homevpn-pc-01-abcdef123456") self.assertEqual(payload["tags"], ["homevpn-pc-01-abcdef123456"[:32]]) self.assertEqual(payload["uses_remaining"], 1) self.assertEqual(payload["networks"], ["office"]) self.assertEqual(headers["Authorization"], "Bearer secret") @override_settings( NETMAKER_BASE_URL="https://netmaker.example.com", NETMAKER_API_TOKEN="secret", NETMAKER_NETWORK_ID="office", NETMAKER_ENROLLMENT_TAGS=["homevpn", "desktop", "vpn"], NETMAKER_ENROLLMENT_TTL_SECONDS=600, ) @patch("vpn.services.netmaker_client.uuid4") @patch("vpn.services.netmaker_client.requests.post") def test_create_one_time_join_token_uses_unique_key_name(self, post, uuid4): first = Mock() first.hex = "111111111111aaaaaaaa" second = Mock() second.hex = "222222222222bbbbbbbb" uuid4.side_effect = [first, second] response = Mock() response.status_code = 200 response.json.return_value = {"token": "join-token"} post.return_value = response NetmakerClient().create_one_time_join_token("pc-01") NetmakerClient().create_one_time_join_token("pc-01") first_payload = post.call_args_list[0].kwargs["json"] second_payload = post.call_args_list[1].kwargs["json"] self.assertNotEqual(first_payload["name"], second_payload["name"]) @override_settings(NETMAKER_BASE_URL="", NETMAKER_API_TOKEN="", NETMAKER_NETWORK_ID="") def test_missing_config_raises(self): with self.assertRaises(NetmakerConfigError): NetmakerClient().create_one_time_join_token("pc-01") @override_settings( NETMAKER_BASE_URL="https://netmaker.example.com", NETMAKER_API_TOKEN="secret", NETMAKER_NETWORK_ID="office", NETMAKER_ENROLLMENT_TAGS=["homevpn", "desktop", "vpn"], NETMAKER_ENROLLMENT_TTL_SECONDS=600, ) @patch("vpn.services.netmaker_client.requests.post") def test_api_error_raises(self, post): response = Mock() response.status_code = 500 response.text = "server error" post.return_value = response with self.assertRaises(NetmakerApiError): NetmakerClient().create_one_time_join_token("pc-01") @override_settings( NETMAKER_BASE_URL="https://netmaker.example.com", NETMAKER_API_TOKEN="secret", NETMAKER_NETWORK_ID="office", ) @patch("vpn.services.netmaker_client.requests.delete") def test_delete_node_uses_host_endpoint(self, delete): response = Mock() response.status_code = 204 response.text = "" delete.return_value = response NetmakerClient().delete_node("host-01") self.assertEqual(delete.call_args.args[0], "https://netmaker.example.com/api/v1/hosts/host-01") self.assertEqual(delete.call_args.kwargs["headers"]["Authorization"], "Bearer secret") @override_settings( NETMAKER_BASE_URL="https://netmaker.example.com", NETMAKER_API_TOKEN="secret", NETMAKER_NETWORK_ID="office", ) @patch("vpn.services.netmaker_client.requests.delete") def test_delete_node_falls_back_to_network_node_endpoint(self, delete): first = Mock() first.status_code = 405 first.text = "method not allowed" second = Mock() second.status_code = 204 second.text = "" delete.side_effect = [first, second] NetmakerClient().delete_node("node-01") urls = [call.args[0] for call in delete.call_args_list] self.assertEqual(urls[0], "https://netmaker.example.com/api/v1/hosts/node-01") self.assertEqual(urls[1], "https://netmaker.example.com/api/v1/nodes/office/node-01") @override_settings(JWT_SECRET="test-secret-key-with-at-least-32-bytes") class VpnApiTests(TestCase): def setUp(self): self.user = User.objects.create(feishu_open_id="open-id", email="u@example.com", name="User") self.jwt = issue_jwt(self.user.id) self.device = Device.objects.create( user=self.user, device_name="pc-01", device_fingerprint="fingerprint-01", ) def test_status_returns_devices_and_logs(self): VpnLog.objects.create(user=self.user, device=self.device, action="connect_token_issued") response = self.client.get("/vpn/status", HTTP_AUTHORIZATION=f"Bearer {self.jwt}") self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["user"]["name"], "User") self.assertEqual(data["user"]["email"], "u@example.com") self.assertEqual(len(data["devices"]), 1) self.assertEqual(data["devices"][0]["device_name"], "pc-01") self.assertEqual(len(data["logs"]), 1) def test_register_node_stores_netmaker_identifier(self): response = self.client.post( "/vpn/register-node", data='{"device_name": "pc-01", "device_fingerprint": "fingerprint-01", "netmaker_node_id": "host-01"}', content_type="application/json", HTTP_AUTHORIZATION=f"Bearer {self.jwt}", ) self.assertEqual(response.status_code, 200) self.device.refresh_from_db() self.assertEqual(self.device.netmaker_node_id, "host-01") self.assertTrue(VpnLog.objects.filter(user=self.user, action="netmaker_node_registered").exists()) def test_revoke_device_marks_device_revoked(self): response = self.client.post( "/vpn/revoke-device", data='{"device_id": %d}' % self.device.id, content_type="application/json", HTTP_AUTHORIZATION=f"Bearer {self.jwt}", ) self.assertEqual(response.status_code, 200) self.device.refresh_from_db() self.assertEqual(self.device.status, Device.STATUS_REVOKED) self.assertTrue(VpnLog.objects.filter(user=self.user, action="device_revoked").exists()) def test_disconnect_log_records_action(self): response = self.client.post( "/vpn/disconnect-log", data='{"device_fingerprint": "fingerprint-01"}', content_type="application/json", HTTP_AUTHORIZATION=f"Bearer {self.jwt}", ) self.assertEqual(response.status_code, 200) log = VpnLog.objects.get(user=self.user, action="disconnect") self.assertEqual(log.device_id, self.device.id)