diff --git a/tests/factories/__init__.py b/tests/factories/__init__.py index de7c412d7..d3fac5caf 100644 --- a/tests/factories/__init__.py +++ b/tests/factories/__init__.py @@ -20,7 +20,12 @@ SwitchPortFactory, ) from .net import SubnetFactory, VLANFactory -from .property import PropertyGroupFactory, MembershipFactory, AdminPropertyGroupFactory +from .property import ( + PropertyGroupFactory, + MembershipFactory, + AdminPropertyGroupFactory, + MemberPropertyGroupFactory, +) from .user import UserFactory, UnixAccountFactory from .traffic import TrafficDataFactory, TrafficVolumeFactory, TrafficVolumeLastWeekFactory from .log import RoomLogEntryFactory, UserLogEntryFactory diff --git a/tests/frontend/test_facilities.py b/tests/frontend/test_facilities.py index 1c6e8a107..45a571709 100644 --- a/tests/frontend/test_facilities.py +++ b/tests/frontend/test_facilities.py @@ -7,6 +7,8 @@ from sqlalchemy.orm import Session from pycroft.model.facilities import Building, Room, Site +from pycroft.model.port import PatchPort +from web.blueprints.facilities.address import ADDRESS_ENTITIES from tests import factories as f from tests.factories import RoomFactory from .assertions import TestClient @@ -53,8 +55,9 @@ def test_get_nonexistent_site(self, client): class TestBuilding: @pytest.fixture(scope="class") - def room(self, class_session: Session) -> Room: + def room(self, class_session, admin) -> Room: room = RoomFactory() + f.RoomLogEntryFactory(room=room, author=admin) class_session.flush() return room @@ -84,6 +87,23 @@ def test_show_nonexistent_room(self, client): url_for("facilities.room_show", room_id=999), code=404 ) + def test_post_room_logs(self, client, room): + with client.renders_template( + "facilities/room_show.html" + ), client.flashes_message("Kommentar hinzugefügt", category="success"): + client.assert_url_ok( + url_for("facilities.room_show", room_id=room.id), + method="POST", + data={"message": "Dose zugeklebt"}, + ) + + def test_room_logs_json(self, client, room): + resp = client.assert_url_ok( + url_for("facilities.room_logs_json", room_id=room.id) + ) + assert "items" in (j := resp.json) + assert j["items"] + def test_building_levels(self, client: TestClient, building: Building): with client.renders_template("facilities/levels.html"): client.assert_url_ok(f"/facilities/building/{building.id}/levels/") @@ -93,23 +113,62 @@ def test_nonexistent_building_levels(self, client): url_for("facilities.building_levels", building_id=999), code=404 ) + def test_building_levels_json(self, client, building): + resp = client.assert_url_ok( + url_for("facilities.json_levels", building=building.id) + ) + assert resp.json.get("items") + + def test_building_rooms_json(self, client, room): + resp = client.assert_url_ok( + url_for( + "facilities.json_rooms", building=room.building.id, level=room.level + ) + ) + assert resp.json.get("items") + def test_building_level_rooms( self, client: TestClient, building: Building, room: Room ): with client.renders_template("facilities/rooms.html"): client.assert_url_ok( - f"/facilities/building/{building.id}/level/{room.level}/rooms/") + f"/facilities/building/{building.id}/level/{room.level}/rooms/" + ) - def test_overcrowded_rooms(self, client: TestClient, building: Building): + +class TestOvercrowdedRooms: + @pytest.fixture(scope="class", autouse=True) + def building(self, class_session) -> Building: + building = f.BuildingFactory() + room = f.RoomFactory(building=building, patched_with_subnet=True) + pg = f.MemberPropertyGroupFactory() + f.UserFactory.create_batch( + 4, + room=room, + with_membership=True, + membership__group=pg, + with_host=True, + ) + return building + + def test_overcrowded_rooms(self, client, building): with client.renders_template("facilities/room_overcrowded.html"): client.assert_url_ok("/facilities/overcrowded") - def test_per_building_overcrowded_rooms( - self, client: TestClient, building: Building - ): + def test_overcrowded_rooms_json(self, client): + resp = client.assert_url_ok(url_for("facilities.overcrowded_json")) + assert len(resp.json.get("items")) == 1 + + def test_per_building_overcrowded_rooms(self, client, building): with client.renders_template("facilities/room_overcrowded.html"): client.assert_url_ok(f"/facilities/overcrowded/{building.id}") + def test_per_building_overcrowded_json(self, client, building): + resp = client.assert_url_ok( + url_for("facilities.overcrowded_json", building=building.id) + ) + assert len(resp.json.get("items")) == 1 + class TestRoomCreate: @pytest.fixture(scope="session") @@ -239,3 +298,248 @@ def test_post_correct_data_ambiguous_name(self, url, client, room, other_room): } with client.renders_template("generic_form.html"): client.assert_url_ok(url, method="POST", data=formdata) + + +class TestBuildingLevelRooms: + @pytest.fixture(scope="class") + def ep(self) -> str: + return "facilities.building_level_rooms_json" + + @pytest.fixture(scope="class", autouse=True) + def building(self, class_session) -> Building: + return f.BuildingFactory() + + @pytest.fixture(scope="class", autouse=True) + def inhabited_room(self, class_session, building) -> Room: + room = f.RoomFactory(level=1, building=building) + group = f.MemberPropertyGroupFactory() + f.UserFactory(room=room, with_membership=True, membership__group=group) + f.UserFactory(room=room) + return room + + @pytest.fixture(scope="class", autouse=True) + def uninhabited_room(self, building): + return f.RoomFactory(level=1, building=building) + + def test_all_users(self, client, ep, building): + url = url_for(ep, building_id=building.id, level=1, all_users=1) + resp = client.assert_url_ok(url) + assert "items" in (j := resp.json) + assert len(j["items"]) == 2 + assert {len(r["inhabitants"]) for r in j["items"]} == {0, 2} + + def test_not_all_users(self, client, ep, building): + url = url_for(ep, building_id=building.id, level=1) + resp = client.assert_url_ok(url) + assert "items" in (j := resp.json) + assert len(j["items"]) == 2 + assert {len(r["inhabitants"]) for r in j["items"]} == {0, 1} + + +class TestPatchPortCreate: + @pytest.fixture(scope="class") + def ep(self) -> str: + return "facilities.patch_port_create" + + @pytest.fixture(scope="class") + def room(self, class_session) -> Room: + return f.RoomFactory() + + @pytest.fixture(scope="class") + def switch_room(self, class_session) -> Room: + switch = f.SwitchFactory() + return switch.host.room + + @pytest.fixture(scope="class") + def patch_port(self, switch_room) -> PatchPort: + return f.PatchPortFactory(switch_room=switch_room) + + @pytest.fixture(scope="class") + def url(self, ep, switch_room) -> str: + """Endpoint URL for the patched room, where a POST makes sense""" + return url_for(ep, switch_room_id=switch_room.id) + + def test_get_nonexistent_room(self, client, ep): + with client.flashes_message("Raum.*nicht gefunden", category="error"): + client.assert_url_redirects(url_for(ep, switch_room_id=999)) + + def test_get_non_switch_room(self, client, ep, room): + with client.flashes_message("kein Switchraum", category="error"): + client.assert_url_redirects(url_for(ep, switch_room_id=room.id)) + + def test_get_switch_room(self, client, url): + with client.renders_template("generic_form.html"): + client.assert_url_ok(url) + + def test_post_no_data(self, client, url): + with client.renders_template("generic_form.html"): + client.assert_url_ok(url, method="POST", data={}) + + def test_post_wrong_data(self, client, url): + data = {"switch": "999"} + with client.renders_template("generic_form.html"): + client.assert_url_ok(url, method="POST", data=data) + + def test_post_existing_patch_port(self, client, url, switch_room, room, patch_port): + data = { + "name": patch_port.name, + "switch_room": switch_room.id, + "building": room.building.id, + "level": room.level, + "room_number": room.number, + } + with client.renders_template("generic_form.html"): + client.assert_url_ok(url, method="POST", data=data) + + def test_post_new_patch_port(self, client, url, patch_port, switch_room, room): + data = { + "name": f"{patch_port.name}-2", + "switch_room": switch_room.id, + "building": room.building.id, + "level": room.level, + "room_number": room.number, + } + with client.flashes_message("erfolgreich erstellt", category="success"): + client.assert_url_redirects(url, method="POST", data=data) + + +class TestPatchPortEdit: + @pytest.fixture(scope="class") + def room(self, class_session) -> Room: + room = f.RoomFactory() + f.SwitchFactory(host__room=room) + return room + + @pytest.fixture(scope="class") + def patch_port(self, class_session, room) -> PatchPort: + patch_port = f.PatchPortFactory(name="A01", switch_room=room) + class_session.flush() + return patch_port + + @pytest.fixture(scope="class") + def patch_port2(self, class_session, room) -> PatchPort: + return f.PatchPortFactory(name="A02", switch_room=room) + + @pytest.fixture(scope="class") + def patch_port_other_room(self) -> PatchPort: + return f.PatchPortFactory() + + @pytest.fixture(scope="class") + def ep(self) -> str: + return "facilities.patch_port_edit" + + @pytest.fixture(scope="class") + def url(self, ep, patch_port, room) -> str: + return url_for(ep, switch_room_id=room.id, patch_port_id=patch_port.id) + + def test_get_nonexistent_patch_port(self, client, ep, room): + with client.flashes_message("nicht gefunden", category="error"): + client.assert_url_redirects( + url_for(ep, switch_room_id=room.id, patch_port_id=999) + ) + + def test_get_patch_port_wrong_room(self, client, ep, room, patch_port_other_room): + with client.flashes_message("ist nicht im .*raum", category="error"): + client.assert_url_redirects( + url_for( + ep, switch_room_id=room.id, patch_port_id=patch_port_other_room.id + ) + ) + + def test_get_correct_patch_port(self, client, url): + with client.renders_template("generic_form.html"): + client.assert_url_ok(url) + + @pytest.mark.parametrize("data", [{}, {"name": "foo"}]) + def test_post_correct_data(self, client, url, data): + with client.flashes_message("erfolgreich bearbeitet", category="success"): + client.assert_url_redirects(url, method="POST", data=data) + + def test_post_wrong_data(self, client, url): + data = {"building": "999"} + with client.renders_template("generic_form.html"): + client.assert_url_ok(url, method="POST", data=data) + + def test_post_existing_patch_port(self, client, url, patch_port2): + data = {"name": patch_port2.name} + with client.renders_template("generic_form.html"): + client.assert_url_ok(url, method="POST", data=data) + + +class TestPatchPortDelete: + @pytest.fixture(scope="class") + def ep(self) -> str: + return "facilities.patch_port_delete" + + @pytest.fixture(scope="class") + def patch_port(self, class_session) -> PatchPort: + switch = f.SwitchFactory() + patch_port = f.PatchPortFactory(switch_room=switch.host.room) + class_session.flush() + return patch_port + + @pytest.fixture(scope="class") + def url(self, ep, patch_port) -> str: + return url_for( + ep, switch_room_id=patch_port.switch_room_id, patch_port_id=patch_port.id + ) + + def test_get(self, client, url): + with client.renders_template("generic_form.html"): + client.assert_url_ok(url) + + def test_post(self, client, url): + with client.flashes_message("erfolgreich gelöscht", category="success"): + client.assert_url_redirects(url, method="POST", data={}) + + +class TestPatchPanelJson: + @pytest.fixture(scope="class") + def switch_room(self, class_session) -> Room: + switch = f.SwitchFactory() + sp = f.SwitchPortFactory(switch=switch) + f.PatchPortFactory(switch_room=switch.host.room, switch_port=sp) + class_session.flush() + return switch.host.room + + @pytest.fixture(scope="class") + def other_room(self) -> Room: + return f.RoomFactory() + + @pytest.fixture(scope="class") + def ep(self) -> str: + return "facilities.room_patchpanel_json" + + def test_get_nonexistent_room(self, client, ep): + client.assert_url_response_code(url_for(ep, room_id=999), 404) + + def test_get_non_switch_room(self, client, ep, other_room): + client.assert_url_response_code(url_for(ep, room_id=other_room.id), 400) + + def test_get_room_patchpanel_json(self, client, ep, switch_room): + resp = client.assert_url_ok(url_for(ep, room_id=switch_room.id)) + assert resp.json.get("items") + + +class TestAddresses: + @pytest.fixture(scope="class", autouse=True) + def addresses(self, class_session): + f.AddressFactory.create_batch(10) + + @pytest.fixture(scope="class") + def ep(self) -> str: + return "facilities.addresses" + + @pytest.mark.parametrize("type", ADDRESS_ENTITIES.keys()) + def test_get_address_completion(self, client, ep, type): + resp = client.assert_url_ok(url_for(ep, type=type)) + assert len(resp.json.get("items")) in range(1, 11) + + @pytest.mark.parametrize("query", ["", "foo"]) + @pytest.mark.parametrize("limit", [None, 5]) + def test_get_address_completion_args(self, client, ep, query, limit): + resp = client.assert_url_ok(url_for(ep, type="city", query=query, limit=limit)) + assert "items" in resp.json + + def test_get_address_invalid_type(self, client, ep): + client.assert_url_response_code(url_for(ep, type="foo"), 404) diff --git a/web/blueprints/facilities/__init__.py b/web/blueprints/facilities/__init__.py index a4dde8362..ab550d7de 100644 --- a/web/blueprints/facilities/__init__.py +++ b/web/blueprints/facilities/__init__.py @@ -10,7 +10,16 @@ """ from collections import defaultdict -from flask import Blueprint, flash, jsonify, render_template, url_for, redirect, request +from flask import ( + Blueprint, + flash, + jsonify, + render_template, + url_for, + redirect, + request, + abort, +) from flask_login import current_user from flask_wtf import FlaskForm as Form from sqlalchemy.orm import joinedload, aliased @@ -38,7 +47,6 @@ from web.blueprints.helpers.log import format_room_log_entry from web.blueprints.helpers.user import user_button from web.blueprints.navigation import BlueprintNavigation -from web.type_utils import abort from .address import get_address_entity, address_entity_search_query from .tables import (BuildingLevelRoomTable, RoomLogTable, SiteTable, RoomOvercrowdedTable, PatchPortTable) @@ -297,18 +305,21 @@ def building_level_rooms_json(level, building_id=None, building_shortname=None): } for room, inhabitants in level_inhabitants.items()]) -@bp.route('/room//patch-port/create', methods=['GET', 'POST']) -@access.require('infrastructure_change') -def patch_port_create(switch_room_id): +def get_switch_room_or_redirect(switch_room_id: int) -> Room: switch_room = Room.get(switch_room_id) - if not switch_room: flash(f"Raum mit ID {switch_room_id} nicht gefunden!", "error") - return redirect(url_for('.overview')) - + abort(redirect(url_for(".overview"))) if not switch_room.is_switch_room: flash("Dieser Raum ist kein Switchraum!", "error") - return redirect(url_for('.room_show', room_id=switch_room_id)) + abort(redirect(url_for(".room_show", room_id=switch_room_id))) + return switch_room + + +@bp.route('/room//patch-port/create', methods=['GET', 'POST']) +@access.require('infrastructure_change') +def patch_port_create(switch_room_id): + switch_room = get_switch_room_or_redirect(switch_room_id) form = PatchPortForm(switch_room=switch_room.short_name, building=switch_room.building, @@ -318,20 +329,22 @@ def patch_port_create(switch_room_id): room = Room.q.filter_by(building=form.building.data, level=form.level.data, number=form.room_number.data).one() + sess = session.session try: - patch_port = create_patch_port(form.name.data, room, switch_room, current_user) - - session.session.commit() - + with sess.begin_nested(): + patch_port = create_patch_port( + form.name.data, room, switch_room, current_user + ) + except PatchPortAlreadyExistsException: + form.name.errors.append( + "Ein Patch-Port mit dieser Bezeichnung existiert bereits in diesem Switchraum." + ) + else: + sess.commit() flash( f"Der Patch-Port {patch_port.name} zum Zimmer {patch_port.room.short_name} wurde erfolgreich erstellt.", "success") - return redirect(url_for('.room_show', room_id=switch_room_id, _anchor="patchpanel")) - except PatchPortAlreadyExistsException: - session.session.rollback() - - form.name.errors.append("Ein Patch-Port mit dieser Bezeichnung existiert bereits in diesem Switchraum.") form_args = { 'form': form, @@ -343,28 +356,24 @@ def patch_port_create(switch_room_id): form_args=form_args) -@bp.route('/room//patch-port//edit', methods=['GET', 'POST']) -@access.require('infrastructure_change') -def patch_port_edit(switch_room_id, patch_port_id): - switch_room = Room.get(switch_room_id) +def get_patch_port_or_redirect( + patch_port_id: int, in_switch_room: Room | None = None +) -> PatchPort: patch_port = PatchPort.get(patch_port_id) - - if not switch_room: - flash(f"Raum mit ID {switch_room_id} nicht gefunden!", "error") - return redirect(url_for('.overview')) - - if not switch_room.is_switch_room: - flash("Dieser Raum ist kein Switchraum!", "error") - return redirect(url_for('.room_show', room_id=switch_room_id)) - if not patch_port: flash(f"Patch-Port mit ID {patch_port_id} nicht gefunden!", "error") - return redirect(url_for('.room_show', room_id=switch_room_id)) - - if not patch_port.switch_room == switch_room: + abort(redirect(url_for(".overview"))) + if in_switch_room and patch_port.switch_room != in_switch_room: flash("Patch-Port ist nicht im Switchraum!", "error") - return redirect(url_for('.room_show', room_id=switch_room_id)) + abort(redirect(url_for(".room_show", room_id=in_switch_room.id))) + return patch_port + +@bp.route('/room//patch-port//edit', methods=['GET', 'POST']) +@access.require('infrastructure_change') +def patch_port_edit(switch_room_id, patch_port_id): + switch_room = get_switch_room_or_redirect(switch_room_id) + patch_port = get_patch_port_or_redirect(patch_port_id, in_switch_room=switch_room) form = PatchPortForm(switch_room=switch_room.short_name, name=patch_port.name, building=patch_port.room.building, @@ -375,19 +384,18 @@ def patch_port_edit(switch_room_id, patch_port_id): room = Room.q.filter_by(building=form.building.data, level=form.level.data, number=form.room_number.data).one() - + sess = session.session try: - edit_patch_port(patch_port, form.name.data, room,current_user) - - session.session.commit() - + with sess.begin_nested(): + edit_patch_port(patch_port, form.name.data, room, current_user) + except PatchPortAlreadyExistsException: + form.name.errors.append( + "Ein Patch-Port mit dieser Bezeichnung existiert bereits in diesem Switchraum." + ) + else: + sess.commit() flash("Der Patch-Port wurde erfolgreich bearbeitet.", "success") - return redirect(url_for('.room_show', room_id=switch_room_id, _anchor="patchpanel")) - except PatchPortAlreadyExistsException: - session.session.rollback() - - form.name.errors.append("Ein Patch-Port mit dieser Bezeichnung existiert bereits in diesem Switchraum.") form_args = { 'form': form, @@ -402,24 +410,8 @@ def patch_port_edit(switch_room_id, patch_port_id): @bp.route('/room//patch-port//delete', methods=['GET', 'POST']) @access.require('infrastructure_change') def patch_port_delete(switch_room_id, patch_port_id): - switch_room = Room.get(switch_room_id) - patch_port = PatchPort.get(patch_port_id) - - if not switch_room: - flash(f"Raum mit ID {switch_room_id} nicht gefunden!", "error") - return redirect(url_for('.overview')) - - if not switch_room.is_switch_room: - flash("Dieser Raum ist kein Switchraum!", "error") - return redirect(url_for('.room_show', room_id=switch_room_id)) - - if not patch_port: - flash(f"Patch-Port mit ID {patch_port_id} nicht gefunden!", "error") - return redirect(url_for('.room_show', room_id=switch_room_id)) - - if not patch_port.switch_room == switch_room: - flash("Patch-Port ist nicht im Switchraum!", "error") - return redirect(url_for('.room_show', room_id=switch_room_id)) + switch_room = get_switch_room_or_redirect(switch_room_id) + patch_port = get_patch_port_or_redirect(patch_port_id, in_switch_room=switch_room) form = Form()