Merge remote-tracking branch 'origin/feat/test'

This commit is contained in:
zuk 2024-10-09 18:45:23 +02:00
commit f5dc8d430f
3 changed files with 98 additions and 4 deletions

View file

@ -1,2 +1,4 @@
[tool.black]
line-length = 160
line-length = 160
[tool.pytest.ini_options]
DJANGO_SETTINGS_MODULE = "wg_manager.settings"

92
test/test.py Normal file
View file

@ -0,0 +1,92 @@
import pytest
from django.urls import reverse
from wg_connection_manager.models import UserConnection
@pytest.fixture
def test_data(django_user_model):
import random
import base64
connections = []
users = []
users.append(django_user_model.objects.create_user(username="user1", password="something"))
users.append(django_user_model.objects.create_user(username="user2", password="something"))
for i in range(10):
key = base64.b64encode(random.randbytes(32))
note = f"connection n: {i}"
connection = UserConnection(public_key=key, notes=note, vpn_ip=f"10.0.0.{i}", user=users[i % 2])
connections.append(connection)
connection.save()
return {"users": users, "connections": connections}
def test_bao(client, admin_client, test_data, mocker):
mock = mocker.patch("wg_connection_manager.views.DJWGManager", autospec=True)
mock.remove_peer = mocker.Mock()
# admin sees all connections
response = admin_client.get(reverse("connections_list"))
assert response.status_code == 200
content = response.content.decode("utf8")
for i, c in enumerate(test_data["connections"]):
assert c.notes in content
# user must authenticate
response = client.get(reverse("connections_list"))
assert response.status_code == 302
# user sees his connection
client.login(username="user1", password="something")
response = client.get(reverse("connections_list"))
assert response.status_code == 200
content = response.content.decode("utf8")
for i, c in enumerate(test_data["connections"]):
if i % 2 == 0:
assert c.notes in content
else:
assert c.notes not in content
# DJWGManager is called to remove peer on connection disable
connection = UserConnection.objects.get(public_key=test_data["connections"][0].public_key)
connection_public_key = connection.public_key
response = client.post(reverse("toggle_connection", args=[connection.id]), {"state": "disabled"})
mock.assert_has_calls([mocker.call(), mocker.call().remove_peer(connection)])
assert response.status_code == 302
# DJWGManager is called to add peer on connection enable
mock.reset_mock()
response = client.post(reverse("toggle_connection", args=[connection.id]), {"state": "enabled"})
mock.assert_has_calls([mocker.call(), mocker.call().add_peer(connection)])
assert response.status_code == 302
# user can delete his connection and DJWGManager is called to remove peer
mock.reset_mock()
# activate connection to call remove_peer
connection.active = True
connection.save()
response = client.post(reverse("connection_delete", args=[connection.id]))
assert response.status_code == 302
# connection is deleted
with pytest.raises(UserConnection.DoesNotExist):
UserConnection.objects.get(pk=1)
# the connection is not in the list anymore
response = admin_client.get(reverse("connections_list"))
assert response.status_code == 200
content = response.content.decode("utf8")
connection_public_key not in content
# user can't delete a connection id doesn't own
mock.reset_mock()
response = client.post(reverse("connection_delete", args=[2]))
mock.assert_not_called()
assert response.status_code == 404
# user can't disable a connection id doesn't own
mock.reset_mock()
response = client.post(reverse("toggle_connection", args=[2]), {"state": "disabled"})
mock.assert_not_called()
assert response.status_code == 404

View file

@ -1,5 +1,5 @@
from django.shortcuts import render, get_object_or_404, redirect
from django.http import HttpRequest, HttpResponseNotAllowed, Http404, HttpResponseBadRequest
from django.http import HttpRequest, HttpResponseNotAllowed, HttpResponseNotFound, HttpResponseBadRequest
from .models import UserConnection
from django.contrib.auth.decorators import login_required
from .dj_wg_manager import DJWGManager
@ -45,7 +45,7 @@ def toggle_connection(request: HttpRequest, connection_id: int):
return HttpResponseBadRequest("invalid state")
connection = get_object_or_404(UserConnection, pk=connection_id)
if connection.user != request.user and not request.user.is_superuser:
return Http404()
return HttpResponseNotFound()
wg_manager = DJWGManager()
match request.POST["state"]:
case "enabled":
@ -59,7 +59,7 @@ def toggle_connection(request: HttpRequest, connection_id: int):
def delete_connection(request: HttpRequest, connection_id: int):
connection = get_object_or_404(UserConnection, pk=connection_id)
if connection.user != request.user and not request.user.is_superuser:
return Http404()
return HttpResponseNotFound()
# sync and disable before delete
wg_manager = DJWGManager()
wg_manager.sync()