2026-04-23 15:41:37 +09:30
|
|
|
"""ZeroTier network access integration tests.
|
|
|
|
|
|
|
|
|
|
Covers network CRUD, device registration, access requests, approvals,
|
|
|
|
|
and membership activation. External ZeroTier API calls are mocked.
|
|
|
|
|
"""
|
|
|
|
|
import pytest
|
|
|
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
|
|
|
|
from tests.integration.client.base import ApiError
|
|
|
|
|
from gatehouse_app.utils.constants import OrganizationRole
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def assert_success(response: dict, message_contains: str = "") -> dict:
|
|
|
|
|
data = response.get("data", {})
|
|
|
|
|
assert response.get("success") is not False, (
|
|
|
|
|
f"Expected success but got error: {response.get('message')}"
|
|
|
|
|
)
|
|
|
|
|
if message_contains:
|
|
|
|
|
assert message_contains.lower() in response.get("message", "").lower()
|
|
|
|
|
return data
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class TestZeroTierNetworkCRUD:
|
|
|
|
|
"""Test ZeroTier network lifecycle."""
|
|
|
|
|
|
|
|
|
|
@patch("gatehouse_app.services.portal_network_service.create_network")
|
|
|
|
|
def test_create_network_positive(self, mock_create_network, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-01 — Create ZeroTier network.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin POST /organizations/<id>/networks with mocked ZT API.
|
|
|
|
|
WHY: Networks are the top-level ZeroTier resource.
|
|
|
|
|
EXPECTED: 201 Created.
|
|
|
|
|
"""
|
|
|
|
|
from gatehouse_app.models.zerotier.portal_network import PortalNetwork
|
|
|
|
|
mock_network = MagicMock()
|
|
|
|
|
mock_network.to_dict.return_value = {"id": "net-123", "name": "Test Network"}
|
|
|
|
|
mock_create_network.return_value = mock_network
|
|
|
|
|
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
result = integration_client.post(
|
|
|
|
|
f"/organizations/{org['id']}/networks",
|
|
|
|
|
data={
|
|
|
|
|
"name": "Test Network",
|
|
|
|
|
"zerotier_network_id": "a84ac5c10a6e4c7e",
|
|
|
|
|
"environment": "development",
|
|
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
assert_success(result)
|
|
|
|
|
|
|
|
|
|
def test_list_networks_positive(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-02 — List networks.
|
|
|
|
|
|
|
|
|
|
WHAT: GET /organizations/<id>/networks.
|
|
|
|
|
WHY: Network overview page uses this endpoint.
|
|
|
|
|
EXPECTED: 200 OK with networks array.
|
|
|
|
|
"""
|
|
|
|
|
user = create_test_user(password="MyPassword123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(user["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=user["email"], password="MyPassword123!")
|
|
|
|
|
result = integration_client.get(f"/organizations/{org['id']}/networks")
|
|
|
|
|
assert_success(result)
|
|
|
|
|
|
|
|
|
|
def test_create_network_non_admin_negative(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-03 — Reject network creation as member.
|
|
|
|
|
|
|
|
|
|
WHAT: Member attempts POST /organizations/<id>/networks.
|
|
|
|
|
WHY: Network management is admin-only.
|
|
|
|
|
EXPECTED: 403 Forbidden.
|
|
|
|
|
"""
|
|
|
|
|
member = create_test_user(password="MemberPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(member["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=member["email"], password="MemberPass123!")
|
|
|
|
|
with pytest.raises(ApiError) as exc_info:
|
|
|
|
|
integration_client.post(
|
|
|
|
|
f"/organizations/{org['id']}/networks",
|
|
|
|
|
data={"name": "Hacked", "zerotier_network_id": "a84ac5c10a6e4c7e"},
|
|
|
|
|
)
|
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class TestZeroTierDeviceManagement:
|
|
|
|
|
"""Test device registration and management."""
|
|
|
|
|
|
|
|
|
|
def test_register_device_positive(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-04 — Register a device.
|
|
|
|
|
|
|
|
|
|
WHAT: POST /organizations/<id>/devices.
|
|
|
|
|
WHY: Devices must be registered before network access.
|
|
|
|
|
EXPECTED: 201 Created.
|
|
|
|
|
"""
|
|
|
|
|
user = create_test_user(password="MyPassword123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(user["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=user["email"], password="MyPassword123!")
|
|
|
|
|
result = integration_client.post(
|
|
|
|
|
f"/organizations/{org['id']}/devices",
|
|
|
|
|
data={
|
|
|
|
|
"node_id": "1234567890",
|
|
|
|
|
"nickname": "Test Device",
|
|
|
|
|
"hostname": "test-device",
|
|
|
|
|
},
|
|
|
|
|
)
|
|
|
|
|
# May succeed or fail depending on ZT config; accept both for now
|
|
|
|
|
assert result.get("success") is not False or result.get("code") in (201, 400, 500)
|
|
|
|
|
|
|
|
|
|
def test_list_devices_positive(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-05 — List devices.
|
|
|
|
|
|
|
|
|
|
WHAT: GET /organizations/<id>/devices.
|
|
|
|
|
WHY: Device management page uses this endpoint.
|
|
|
|
|
EXPECTED: 200 OK.
|
|
|
|
|
"""
|
|
|
|
|
user = create_test_user(password="MyPassword123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(user["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=user["email"], password="MyPassword123!")
|
|
|
|
|
result = integration_client.get(f"/organizations/{org['id']}/devices")
|
|
|
|
|
assert_success(result)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class TestZeroTierApprovals:
|
|
|
|
|
"""Test approval flows."""
|
|
|
|
|
|
|
|
|
|
def test_list_pending_approvals_positive(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-06 — List pending approvals as admin.
|
|
|
|
|
|
|
|
|
|
WHAT: GET /organizations/<id>/approvals/pending.
|
|
|
|
|
WHY: Admins review pending access requests.
|
|
|
|
|
EXPECTED: 200 OK.
|
|
|
|
|
"""
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
result = integration_client.get(f"/organizations/{org['id']}/approvals/pending")
|
|
|
|
|
assert_success(result)
|
|
|
|
|
|
|
|
|
|
def test_list_approvals_positive(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-07 — List all approvals.
|
|
|
|
|
|
|
|
|
|
WHAT: GET /organizations/<id>/approvals.
|
|
|
|
|
WHY: Approval history page uses this endpoint.
|
|
|
|
|
EXPECTED: 200 OK.
|
|
|
|
|
"""
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
result = integration_client.get(f"/organizations/{org['id']}/approvals")
|
|
|
|
|
assert_success(result)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class TestZeroTierMembership:
|
|
|
|
|
"""Test membership activation and deactivation."""
|
|
|
|
|
|
|
|
|
|
def test_get_memberships_positive(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-08 — Get ZeroTier memberships.
|
|
|
|
|
|
|
|
|
|
WHAT: GET /organizations/<id>/memberships.
|
|
|
|
|
WHY: Users see their active network memberships.
|
|
|
|
|
EXPECTED: 200 OK.
|
|
|
|
|
"""
|
|
|
|
|
user = create_test_user(password="MyPassword123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(user["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=user["email"], password="MyPassword123!")
|
|
|
|
|
result = integration_client.get(f"/organizations/{org['id']}/memberships")
|
|
|
|
|
assert_success(result)
|
|
|
|
|
|
|
|
|
|
def test_kill_switch_positive(self, integration_client, create_test_user, create_test_org, create_test_membership):
|
|
|
|
|
"""TEST: ZT-09 — Trigger kill switch.
|
|
|
|
|
|
|
|
|
|
WHAT: POST /organizations/<id>/kill-switch.
|
|
|
|
|
WHY: Emergency access revocation.
|
|
|
|
|
EXPECTED: 200 OK or error if no memberships exist.
|
|
|
|
|
"""
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
try:
|
|
|
|
|
result = integration_client.post(
|
|
|
|
|
f"/organizations/{org['id']}/kill-switch",
|
|
|
|
|
data={"target_user_id": admin["id"], "reason": "Test kill switch"},
|
|
|
|
|
)
|
|
|
|
|
assert_success(result)
|
|
|
|
|
except ApiError as exc:
|
|
|
|
|
# Accept errors when no active memberships to kill
|
|
|
|
|
assert exc.status_code in (400, 500)
|
2026-05-07 20:04:08 +00:00
|
|
|
|
|
|
|
|
|
2026-05-28 05:42:04 +00:00
|
|
|
class TestZeroTierJoinNetwork:
|
|
|
|
|
"""Test joining a network with a registered device."""
|
|
|
|
|
|
|
|
|
|
@patch("gatehouse_app.services.network_access_service._ensure_zerotier_member")
|
|
|
|
|
def test_rejoin_after_deactivation_positive(
|
|
|
|
|
self,
|
|
|
|
|
mock_ensure_member,
|
|
|
|
|
integration_client,
|
|
|
|
|
create_test_user,
|
|
|
|
|
create_test_org,
|
|
|
|
|
create_test_membership,
|
|
|
|
|
integration_app,
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-15 — Re-join network after deactivation.
|
|
|
|
|
|
|
|
|
|
WHAT: User with deactivated membership (APPROVED + active=False)
|
|
|
|
|
POSTs to join-network for the same device and network.
|
|
|
|
|
WHY: Deactivated memberships should not block re-join attempts.
|
|
|
|
|
EXPECTED: 201 Created (re-opens existing request), not 409 Conflict.
|
|
|
|
|
"""
|
|
|
|
|
from gatehouse_app.models.zerotier.device import Device
|
|
|
|
|
from gatehouse_app.models.zerotier.portal_network import PortalNetwork
|
|
|
|
|
from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest
|
|
|
|
|
from gatehouse_app.extensions import db as _db
|
|
|
|
|
from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode
|
|
|
|
|
|
|
|
|
|
user = create_test_user(password="MyPassword123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(user["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
with integration_app.app_context():
|
|
|
|
|
device = Device(
|
|
|
|
|
user_id=user["id"],
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
node_id="1234567890",
|
|
|
|
|
device_nickname="Test Device",
|
|
|
|
|
hostname="test-device",
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(device)
|
|
|
|
|
_db.session.flush()
|
|
|
|
|
device_id = device.id
|
|
|
|
|
|
|
|
|
|
portal_network = PortalNetwork(
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
name="Test Network",
|
|
|
|
|
owner_user_id=user["id"],
|
|
|
|
|
zerotier_network_id="a84ac5c10a6e4c7e",
|
|
|
|
|
request_mode=NetworkRequestMode.OPEN,
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(portal_network)
|
|
|
|
|
_db.session.flush()
|
|
|
|
|
portal_network_id = portal_network.id
|
|
|
|
|
|
|
|
|
|
deactivated_request = NetworkAccessRequest(
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
user_id=user["id"],
|
|
|
|
|
device_id=device_id,
|
|
|
|
|
portal_network_id=portal_network_id,
|
|
|
|
|
status=ApprovalState.APPROVED,
|
|
|
|
|
active=False,
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(deactivated_request)
|
|
|
|
|
_db.session.commit()
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=user["email"], password="MyPassword123!")
|
|
|
|
|
result = integration_client.post(
|
|
|
|
|
f"/organizations/{org['id']}/devices/{device_id}/join-network/{portal_network_id}",
|
|
|
|
|
)
|
|
|
|
|
data = assert_success(result, "joined network successfully")
|
|
|
|
|
assert result.get("code") == 201
|
|
|
|
|
assert "membership" in data
|
|
|
|
|
mock_ensure_member.assert_called_once()
|
|
|
|
|
|
|
|
|
|
|
2026-05-07 20:04:08 +00:00
|
|
|
class TestAdminUserDevices:
|
|
|
|
|
"""Test admin endpoint to list devices for a specific user."""
|
|
|
|
|
|
|
|
|
|
def test_list_user_devices_positive(
|
|
|
|
|
self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-10 — Admin lists devices for a user with devices.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin GET /organizations/<id>/users/<user_id>/devices.
|
|
|
|
|
WHY: Admins need to see what devices a user has registered.
|
|
|
|
|
EXPECTED: 200 OK with devices array.
|
|
|
|
|
"""
|
|
|
|
|
from gatehouse_app.models.zerotier.device import Device
|
|
|
|
|
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
member = create_test_user(password="MemberPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
create_test_membership(member["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
# Create test devices for the member
|
|
|
|
|
from gatehouse_app.extensions import db as _db
|
|
|
|
|
with integration_app.app_context():
|
|
|
|
|
device1 = Device(
|
|
|
|
|
user_id=member["id"],
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
node_id="1234567890",
|
|
|
|
|
device_nickname="Member Laptop",
|
|
|
|
|
hostname="member-laptop",
|
|
|
|
|
)
|
|
|
|
|
device2 = Device(
|
|
|
|
|
user_id=member["id"],
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
node_id="0987654321",
|
|
|
|
|
device_nickname="Member Phone",
|
|
|
|
|
hostname="member-phone",
|
|
|
|
|
)
|
|
|
|
|
_db.session.add_all([device1, device2])
|
|
|
|
|
_db.session.commit()
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
result = integration_client.get(f"/organizations/{org['id']}/users/{member['id']}/devices")
|
|
|
|
|
data = assert_success(result, "devices retrieved")
|
|
|
|
|
|
|
|
|
|
assert "devices" in data
|
|
|
|
|
assert data["count"] == 2
|
|
|
|
|
assert data["user_id"] == member["id"]
|
|
|
|
|
assert data["organization_id"] == org["id"]
|
|
|
|
|
device_node_ids = [d["node_id"] for d in data["devices"]]
|
|
|
|
|
assert "1234567890" in device_node_ids
|
|
|
|
|
assert "0987654321" in device_node_ids
|
|
|
|
|
|
|
|
|
|
def test_list_user_devices_no_devices(
|
|
|
|
|
self, integration_client, create_test_user, create_test_org, create_test_membership
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-11 — Admin lists devices for a user with no devices.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin GET /organizations/<id>/users/<user_id>/devices for user with no devices.
|
|
|
|
|
WHY: Endpoint should return empty list, not error.
|
|
|
|
|
EXPECTED: 200 OK with empty devices array.
|
|
|
|
|
"""
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
member = create_test_user(password="MemberPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
create_test_membership(member["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
result = integration_client.get(f"/organizations/{org['id']}/users/{member['id']}/devices")
|
|
|
|
|
data = assert_success(result)
|
|
|
|
|
|
|
|
|
|
assert data["count"] == 0
|
|
|
|
|
assert data["devices"] == []
|
|
|
|
|
|
|
|
|
|
def test_list_user_devices_non_admin_negative(
|
|
|
|
|
self, integration_client, create_test_user, create_test_org, create_test_membership
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-12 — Non-admin cannot list another user's devices.
|
|
|
|
|
|
|
|
|
|
WHAT: Member attempts GET /organizations/<id>/users/<user_id>/devices.
|
|
|
|
|
WHY: This endpoint is admin-only.
|
|
|
|
|
EXPECTED: 403 Forbidden.
|
|
|
|
|
"""
|
|
|
|
|
member1 = create_test_user(password="Member1Pass123!")
|
|
|
|
|
member2 = create_test_user(password="Member2Pass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
|
|
|
|
|
create_test_membership(member1["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
create_test_membership(member2["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=member1["email"], password="Member1Pass123!")
|
|
|
|
|
with pytest.raises(ApiError) as exc_info:
|
|
|
|
|
integration_client.get(f"/organizations/{org['id']}/users/{member2['id']}/devices")
|
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
|
|
|
|
def test_list_user_devices_user_not_in_org_negative(
|
|
|
|
|
self, integration_client, create_test_user, create_test_org, create_test_membership
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-13 — Cannot list devices for user not in organization.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin GET /organizations/<id>/users/<user_id>/devices for user not in org.
|
|
|
|
|
WHY: User must be a member of the organization.
|
|
|
|
|
EXPECTED: 404 Not Found.
|
|
|
|
|
"""
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
outside_user = create_test_user(password="OutsidePass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
# outside_user is NOT added to the org
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
with pytest.raises(ApiError) as exc_info:
|
|
|
|
|
integration_client.get(f"/organizations/{org['id']}/users/{outside_user['id']}/devices")
|
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
|
|
|
|
def test_list_user_devices_user_not_found_negative(
|
|
|
|
|
self, integration_client, create_test_user, create_test_org, create_test_membership
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-14 — Cannot list devices for non-existent user.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin GET /organizations/<id>/users/<non_existent_id>/devices.
|
|
|
|
|
WHY: User must exist.
|
|
|
|
|
EXPECTED: 404 Not Found.
|
|
|
|
|
"""
|
|
|
|
|
import uuid
|
|
|
|
|
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
|
|
|
|
|
non_existent_id = str(uuid.uuid4())
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
with pytest.raises(ApiError) as exc_info:
|
|
|
|
|
integration_client.get(f"/organizations/{org['id']}/users/{non_existent_id}/devices")
|
|
|
|
|
assert exc_info.value.status_code == 404
|
2026-05-28 05:42:04 +00:00
|
|
|
|
|
|
|
|
|
|
|
|
|
class TestAdminForceDeleteMembership:
|
|
|
|
|
"""Test admin force-delete of network access requests."""
|
|
|
|
|
|
|
|
|
|
@patch("gatehouse_app.services.zerotier_api_service.deauthorize_member")
|
|
|
|
|
@patch("gatehouse_app.services.zerotier_api_service.delete_network_member")
|
|
|
|
|
def test_force_delete_active_membership_positive(
|
|
|
|
|
self,
|
|
|
|
|
mock_delete_member,
|
|
|
|
|
mock_deauthorize_member,
|
|
|
|
|
integration_client,
|
|
|
|
|
create_test_user,
|
|
|
|
|
create_test_org,
|
|
|
|
|
create_test_membership,
|
|
|
|
|
integration_app,
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-16 — Admin force-deletes an active membership.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin calls DELETE admin/memberships/<id> on an active,
|
|
|
|
|
non-soft-deleted request. Should deactivate, remove from ZT,
|
|
|
|
|
and hard-delete the DB record in one step.
|
|
|
|
|
EXPECTED: 200 OK, request no longer exists in DB.
|
|
|
|
|
"""
|
|
|
|
|
from gatehouse_app.models.zerotier.device import Device
|
|
|
|
|
from gatehouse_app.models.zerotier.portal_network import PortalNetwork
|
|
|
|
|
from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest
|
|
|
|
|
from gatehouse_app.extensions import db as _db
|
|
|
|
|
from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode
|
|
|
|
|
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
user = create_test_user(password="UserPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
create_test_membership(user["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
with integration_app.app_context():
|
|
|
|
|
device = Device(
|
|
|
|
|
user_id=user["id"],
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
node_id="1234567890",
|
|
|
|
|
device_nickname="Test Device",
|
|
|
|
|
hostname="test-device",
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(device)
|
|
|
|
|
_db.session.flush()
|
|
|
|
|
device_id = device.id
|
|
|
|
|
|
|
|
|
|
portal_network = PortalNetwork(
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
name="Test Network",
|
|
|
|
|
owner_user_id=admin["id"],
|
|
|
|
|
zerotier_network_id="a84ac5c10a6e4c7e",
|
|
|
|
|
request_mode=NetworkRequestMode.OPEN,
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(portal_network)
|
|
|
|
|
_db.session.flush()
|
|
|
|
|
portal_network_id = portal_network.id
|
|
|
|
|
|
|
|
|
|
request = NetworkAccessRequest(
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
user_id=user["id"],
|
|
|
|
|
device_id=device_id,
|
|
|
|
|
portal_network_id=portal_network_id,
|
|
|
|
|
status=ApprovalState.APPROVED,
|
|
|
|
|
active=True,
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(request)
|
|
|
|
|
_db.session.commit()
|
|
|
|
|
request_id = request.id
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
result = integration_client.delete(
|
|
|
|
|
f"/organizations/{org['id']}/admin/memberships/{request_id}",
|
|
|
|
|
)
|
|
|
|
|
assert_success(result, "permanently deleted")
|
|
|
|
|
|
|
|
|
|
# Verify request is gone from DB
|
|
|
|
|
with integration_app.app_context():
|
|
|
|
|
deleted = NetworkAccessRequest.query.get(request_id)
|
|
|
|
|
assert deleted is None
|
|
|
|
|
|
|
|
|
|
mock_deauthorize_member.assert_called_once()
|
|
|
|
|
mock_delete_member.assert_called_once()
|
|
|
|
|
|
|
|
|
|
@patch("gatehouse_app.services.zerotier_api_service.delete_network_member")
|
|
|
|
|
def test_force_delete_soft_deleted_membership_positive(
|
|
|
|
|
self,
|
|
|
|
|
mock_delete_member,
|
|
|
|
|
integration_client,
|
|
|
|
|
create_test_user,
|
|
|
|
|
create_test_org,
|
|
|
|
|
create_test_membership,
|
|
|
|
|
integration_app,
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-17 — Admin force-deletes an already-soft-deleted membership.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin calls DELETE admin/memberships/<id> on a soft-deleted
|
|
|
|
|
request. Should still work (remove from ZT, hard-delete).
|
|
|
|
|
EXPECTED: 200 OK.
|
|
|
|
|
"""
|
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
|
from gatehouse_app.models.zerotier.device import Device
|
|
|
|
|
from gatehouse_app.models.zerotier.portal_network import PortalNetwork
|
|
|
|
|
from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest
|
|
|
|
|
from gatehouse_app.extensions import db as _db
|
|
|
|
|
from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode
|
|
|
|
|
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
user = create_test_user(password="UserPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
create_test_membership(user["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
with integration_app.app_context():
|
|
|
|
|
device = Device(
|
|
|
|
|
user_id=user["id"],
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
node_id="1234567890",
|
|
|
|
|
device_nickname="Test Device",
|
|
|
|
|
hostname="test-device",
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(device)
|
|
|
|
|
_db.session.flush()
|
|
|
|
|
device_id = device.id
|
|
|
|
|
|
|
|
|
|
portal_network = PortalNetwork(
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
name="Test Network",
|
|
|
|
|
owner_user_id=admin["id"],
|
|
|
|
|
zerotier_network_id="a84ac5c10a6e4c7e",
|
|
|
|
|
request_mode=NetworkRequestMode.OPEN,
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(portal_network)
|
|
|
|
|
_db.session.flush()
|
|
|
|
|
portal_network_id = portal_network.id
|
|
|
|
|
|
|
|
|
|
request = NetworkAccessRequest(
|
|
|
|
|
organization_id=org["id"],
|
|
|
|
|
user_id=user["id"],
|
|
|
|
|
device_id=device_id,
|
|
|
|
|
portal_network_id=portal_network_id,
|
|
|
|
|
status=ApprovalState.APPROVED,
|
|
|
|
|
active=False,
|
|
|
|
|
deleted_at=datetime.now(timezone.utc),
|
|
|
|
|
)
|
|
|
|
|
_db.session.add(request)
|
|
|
|
|
_db.session.commit()
|
|
|
|
|
request_id = request.id
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
result = integration_client.delete(
|
|
|
|
|
f"/organizations/{org['id']}/admin/memberships/{request_id}",
|
|
|
|
|
)
|
|
|
|
|
assert_success(result, "permanently deleted")
|
|
|
|
|
|
|
|
|
|
# Verify request is gone from DB
|
|
|
|
|
with integration_app.app_context():
|
|
|
|
|
deleted = NetworkAccessRequest.query.get(request_id)
|
|
|
|
|
assert deleted is None
|
|
|
|
|
|
|
|
|
|
mock_delete_member.assert_called_once()
|
|
|
|
|
|
|
|
|
|
def test_force_delete_non_existent_membership_negative(
|
|
|
|
|
self,
|
|
|
|
|
integration_client,
|
|
|
|
|
create_test_user,
|
|
|
|
|
create_test_org,
|
|
|
|
|
create_test_membership,
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-18 — Admin force-deletes a non-existent membership.
|
|
|
|
|
|
|
|
|
|
WHAT: Admin calls DELETE admin/memberships/<non_existent_id>.
|
|
|
|
|
EXPECTED: 404 Not Found.
|
|
|
|
|
"""
|
|
|
|
|
import uuid
|
|
|
|
|
|
|
|
|
|
admin = create_test_user(password="AdminPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(admin["id"], org["id"], OrganizationRole.ADMIN)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=admin["email"], password="AdminPass123!")
|
|
|
|
|
with pytest.raises(ApiError) as exc_info:
|
|
|
|
|
integration_client.delete(
|
|
|
|
|
f"/organizations/{org['id']}/admin/memberships/{uuid.uuid4()}",
|
|
|
|
|
)
|
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
|
|
|
|
def test_force_delete_non_admin_negative(
|
|
|
|
|
self,
|
|
|
|
|
integration_client,
|
|
|
|
|
create_test_user,
|
|
|
|
|
create_test_org,
|
|
|
|
|
create_test_membership,
|
|
|
|
|
):
|
|
|
|
|
"""TEST: ZT-19 — Non-admin cannot force-delete a membership.
|
|
|
|
|
|
|
|
|
|
WHAT: A MEMBER calls DELETE admin/memberships/<id>.
|
|
|
|
|
EXPECTED: 403 Forbidden.
|
|
|
|
|
"""
|
|
|
|
|
import uuid
|
|
|
|
|
|
|
|
|
|
member = create_test_user(password="MemberPass123!")
|
|
|
|
|
org = create_test_org()
|
|
|
|
|
create_test_membership(member["id"], org["id"], OrganizationRole.MEMBER)
|
|
|
|
|
|
|
|
|
|
integration_client.auth.login(email=member["email"], password="MemberPass123!")
|
|
|
|
|
with pytest.raises(ApiError) as exc_info:
|
|
|
|
|
integration_client.delete(
|
|
|
|
|
f"/organizations/{org['id']}/admin/memberships/{uuid.uuid4()}",
|
|
|
|
|
)
|
|
|
|
|
assert exc_info.value.status_code == 403
|
