Passed
Push — main ( 7dc7dc...65a384 )
by Jochen
03:23
created

occupy_seat()   A

Complexity

Conditions 4

Size

Total Lines 34
Code Lines 21

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 17
CRAP Score 4

Importance

Changes 0
Metric Value
cc 4
eloc 21
nop 3
dl 0
loc 34
ccs 17
cts 17
cp 1
crap 4
rs 9.376
c 0
b 0
f 0
1
"""
2
byceps.services.ticketing.ticket_seat_management_service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2020 Jochen Kupperschmidt
6
:License: Modified BSD, see LICENSE for details.
7
"""
8
9 1
from ...database import db
10 1
from ...typing import UserID
11
12 1
from ..seating.models.seat import Seat as DbSeat
13 1
from ..seating import seat_service
14 1
from ..seating.transfer.models import SeatID
15
16 1
from . import event_service
17 1
from .exceptions import (
18
    SeatChangeDeniedForBundledTicket,
19
    SeatChangeDeniedForGroupSeat,
20
    TicketCategoryMismatch,
21
    TicketIsRevoked,
22
)
23 1
from .models.ticket import Ticket as DbTicket
24 1
from . import ticket_service
25 1
from .transfer.models import TicketID
26
27
28 1
def appoint_seat_manager(
29
    ticket_id: TicketID, manager_id: UserID, initiator_id: UserID
30
) -> None:
31
    """Appoint the user as the ticket's seat manager."""
32 1
    ticket = _get_ticket(ticket_id)
33
34 1
    ticket.seat_managed_by_id = manager_id
35
36 1
    event = event_service.build_event(
37
        'seat-manager-appointed',
38
        ticket.id,
39
        {
40
            'appointed_seat_manager_id': str(manager_id),
41
            'initiator_id': str(initiator_id),
42
        },
43
    )
44 1
    db.session.add(event)
45
46 1
    db.session.commit()
47
48
49 1
def withdraw_seat_manager(ticket_id: TicketID, initiator_id: UserID) -> None:
50
    """Withdraw the ticket's custom seat manager."""
51 1
    ticket = _get_ticket(ticket_id)
52
53 1
    ticket.seat_managed_by_id = None
54
55 1
    event = event_service.build_event(
56
        'seat-manager-withdrawn',
57
        ticket.id,
58
        {
59
            'initiator_id': str(initiator_id),
60
        },
61
    )
62 1
    db.session.add(event)
63
64 1
    db.session.commit()
65
66
67 1
def occupy_seat(
68
    ticket_id: TicketID, seat_id: SeatID, initiator_id: UserID
69
) -> None:
70
    """Occupy the seat with this ticket."""
71 1
    ticket = _get_ticket(ticket_id)
72
73 1
    _deny_seat_management_if_ticket_belongs_to_bundle(ticket)
74
75 1
    seat = seat_service.find_seat(seat_id)
76 1
    if seat is None:
77 1
        raise ValueError('Invalid seat ID')
78
79 1
    if seat.category_id != ticket.category_id:
80 1
        raise TicketCategoryMismatch(
81
            'Ticket and seat belong to different categories.'
82
        )
83
84 1
    _deny_seat_management_if_seat_belongs_to_group(seat)
85
86 1
    previous_seat_id = ticket.occupied_seat_id
87
88 1
    ticket.occupied_seat_id = seat.id
89
90 1
    event_data = {
91
        'seat_id': str(seat.id),
92
        'initiator_id': str(initiator_id),
93
    }
94 1
    if previous_seat_id is not None:
95 1
        event_data['previous_seat_id'] = str(previous_seat_id)
96
97 1
    event = event_service.build_event('seat-occupied', ticket.id, event_data)
98 1
    db.session.add(event)
99
100 1
    db.session.commit()
101
102
103 1
def release_seat(ticket_id: TicketID, initiator_id: UserID) -> None:
104
    """Release the seat occupied by this ticket."""
105 1
    ticket = _get_ticket(ticket_id)
106
107 1
    _deny_seat_management_if_ticket_belongs_to_bundle(ticket)
108
109 1
    seat = seat_service.find_seat(ticket.occupied_seat_id)
110 1
    if seat is None:
111
        raise ValueError('Ticket does not occupy a seat.')
112
113 1
    _deny_seat_management_if_seat_belongs_to_group(seat)
114
115 1
    ticket.occupied_seat_id = None
116
117 1
    event = event_service.build_event(
118
        'seat-released',
119
        ticket.id,
120
        {
121
            'seat_id': str(seat.id),
122
            'initiator_id': str(initiator_id),
123
        },
124
    )
125 1
    db.session.add(event)
126
127 1
    db.session.commit()
128
129
130 1
def _get_ticket(ticket_id: TicketID) -> DbTicket:
131
    """Return the ticket with that ID.
132
133
    Raise an exception if the ID is unknown or if the ticket has been
134
    revoked.
135
    """
136 1
    ticket = ticket_service.find_ticket(ticket_id)
137
138 1
    if ticket is None:
139
        raise ValueError(f'Unknown ticket ID "{ticket_id}"')
140
141 1
    if ticket.revoked:
142
        raise TicketIsRevoked(f'Ticket {ticket_id} has been revoked.')
143
144 1
    return ticket
145
146
147 1
def _deny_seat_management_if_ticket_belongs_to_bundle(ticket: DbTicket) -> None:
148
    """Raise an exception if this ticket belongs to a bundle.
149
150
    A ticket bundle is meant to occupy a matching seat group with the
151
    appropriate mechanism, not to separately occupy single seats.
152
    """
153 1
    if ticket.belongs_to_bundle:
154 1
        raise SeatChangeDeniedForBundledTicket(
155
            f"Ticket '{ticket.code}' belongs to a bundle and, thus, "
156
            'must not be used to occupy or release a single seat.'
157
        )
158
159
160 1
def _deny_seat_management_if_seat_belongs_to_group(seat: DbSeat) -> None:
161 1
    if seat.assignment is not None:
162
        raise SeatChangeDeniedForGroupSeat(
163
            f"Seat '{seat.label}' belongs to a group and, thus, "
164
            'cannot be occupied by a single ticket, or removed separately.'
165
        )
166