11"""Test utilities for OCPI 2.3.0 Bookings module."""
22
3+ import copy
34from datetime import UTC , datetime , timedelta
45
56from ocpi .core .adapter import BaseAdapter
1819CONNECTOR_ID = "CONN-001"
1920EMSP_BOOKING_ID = "EMSP-BOOKING-001"
2021
21- TOKEN_DATA = {
22- "country_code" : "DE" ,
23- "party_id" : "ELU" ,
24- "uid" : "TOKEN-001" ,
25- "type" : TokenType .rfid ,
26- "contract_id" : "CONTRACT-001" ,
27- "visual_number" : "12345678" ,
28- "issuer" : "ELU Mobility" ,
29- "group_id" : None ,
30- "valid" : True ,
31- "whitelist" : WhitelistType .always ,
32- "language" : "de" ,
33- "default_profile_type" : None ,
34- "energy_contract" : None ,
35- "last_updated" : datetime .now (UTC ).isoformat (),
36- }
37-
38- BOOKINGS = [
39- {
22+
23+ def get_token_data ():
24+ """Return fresh token data for each test."""
25+ return {
4026 "country_code" : "DE" ,
4127 "party_id" : "ELU" ,
42- "id" : BOOKING_ID ,
43- "emsp_booking_id" : EMSP_BOOKING_ID ,
44- "token" : TOKEN_DATA ,
45- "location_id" : LOCATION_ID ,
46- "evse_uid" : EVSE_UID ,
47- "connector_id" : CONNECTOR_ID ,
48- "start_date_time" : (datetime .now (UTC ) + timedelta (hours = 1 )).isoformat (),
49- "end_date_time" : (datetime .now (UTC ) + timedelta (hours = 3 )).isoformat (),
50- "state" : BookingState .confirmed ,
51- "authorization_reference" : "AUTH-REF-001" ,
52- "energy_estimate" : 50.0 ,
53- "estimated_cost" : {"excl_vat" : 25.00 , "incl_vat" : 29.75 },
54- "status_message" : [],
55- "session_id" : None ,
28+ "uid" : "TOKEN-001" ,
29+ "type" : TokenType .rfid ,
30+ "contract_id" : "CONTRACT-001" ,
31+ "visual_number" : "12345678" ,
32+ "issuer" : "ELU Mobility" ,
33+ "group_id" : None ,
34+ "valid" : True ,
35+ "whitelist" : WhitelistType .always ,
36+ "language" : "de" ,
37+ "default_profile_type" : None ,
38+ "energy_contract" : None ,
5639 "last_updated" : datetime .now (UTC ).isoformat (),
5740 }
58- ]
41+
42+
43+ def get_bookings ():
44+ """Return fresh booking data for each test - avoids test isolation issues."""
45+ return [
46+ {
47+ "country_code" : "DE" ,
48+ "party_id" : "ELU" ,
49+ "id" : BOOKING_ID ,
50+ "emsp_booking_id" : EMSP_BOOKING_ID ,
51+ "token" : get_token_data (),
52+ "location_id" : LOCATION_ID ,
53+ "evse_uid" : EVSE_UID ,
54+ "connector_id" : CONNECTOR_ID ,
55+ "start_date_time" : (datetime .now (UTC ) + timedelta (hours = 1 )).isoformat (),
56+ "end_date_time" : (datetime .now (UTC ) + timedelta (hours = 3 )).isoformat (),
57+ "state" : BookingState .confirmed ,
58+ "authorization_reference" : "AUTH-REF-001" ,
59+ "energy_estimate" : 50.0 ,
60+ "estimated_cost" : {"excl_vat" : 25.00 , "incl_vat" : 29.75 },
61+ "status_message" : [],
62+ "session_id" : None ,
63+ "last_updated" : datetime .now (UTC ).isoformat (),
64+ }
65+ ]
66+
67+
68+ # Keep for backward compatibility but use fresh copy
69+ TOKEN_DATA = get_token_data ()
70+ BOOKINGS = get_bookings ()
5971
6072
6173class Crud (Crud ):
74+ # Class-level storage that resets for each test class
75+ _bookings = None
76+
77+ @classmethod
78+ def _get_bookings (cls ):
79+ """Get bookings, initializing fresh data if needed."""
80+ if cls ._bookings is None :
81+ cls ._bookings = get_bookings ()
82+ return cls ._bookings
83+
84+ @classmethod
85+ def _reset_bookings (cls ):
86+ """Reset bookings to fresh state."""
87+ cls ._bookings = get_bookings ()
88+
6289 @classmethod
6390 async def list (
6491 cls ,
@@ -68,7 +95,8 @@ async def list(
6895 * args ,
6996 ** kwargs ,
7097 ) -> tuple [list , int , bool ]:
71- return BOOKINGS , len (BOOKINGS ), True
98+ bookings = cls ._get_bookings ()
99+ return bookings , len (bookings ), True
72100
73101 @classmethod
74102 async def get (
@@ -79,9 +107,10 @@ async def get(
79107 * args ,
80108 ** kwargs ,
81109 ):
82- for booking in BOOKINGS :
110+ bookings = cls ._get_bookings ()
111+ for booking in bookings :
83112 if booking ["id" ] == id :
84- return booking
113+ return copy . deepcopy ( booking )
85114 return None
86115
87116 @classmethod
@@ -93,11 +122,12 @@ async def create(
93122 * args ,
94123 ** kwargs ,
95124 ):
125+ bookings = cls ._get_bookings ()
96126 # Create a new booking from request data
97127 new_booking = {
98128 "country_code" : "DE" ,
99129 "party_id" : "ELU" ,
100- "id" : f"BOOKING-{ len (BOOKINGS ) + 1 :03d} " ,
130+ "id" : f"BOOKING-{ len (bookings ) + 1 :03d} " ,
101131 "emsp_booking_id" : data .get ("emsp_booking_id" ),
102132 "token" : data .get ("token" ),
103133 "location_id" : data .get ("location_id" ),
@@ -125,26 +155,29 @@ async def update(
125155 * args ,
126156 ** kwargs ,
127157 ):
128- # For EMSP, also check country_code and party_id if provided
158+ bookings = cls . _get_bookings ()
129159 country_code = kwargs .get ("country_code" )
130160 party_id = kwargs .get ("party_id" )
131161
132- for booking in BOOKINGS :
162+ for booking in bookings :
133163 if booking ["id" ] == id :
134164 # If country_code and party_id are provided, verify they match
135165 if country_code and party_id :
136166 if (
137167 booking ["country_code" ] == country_code
138168 and booking ["party_id" ] == party_id
139169 ):
140- booking .update (data )
141- booking ["last_updated" ] = datetime .now (UTC ).isoformat ()
142- return booking
170+ # Return updated copy without mutating original
171+ updated = copy .deepcopy (booking )
172+ updated .update (data )
173+ updated ["last_updated" ] = datetime .now (UTC ).isoformat ()
174+ return updated
143175 else :
144- # No country_code/party_id check (CPO context)
145- booking .update (data )
146- booking ["last_updated" ] = datetime .now (UTC ).isoformat ()
147- return booking
176+ # Return updated copy without mutating original
177+ updated = copy .deepcopy (booking )
178+ updated .update (data )
179+ updated ["last_updated" ] = datetime .now (UTC ).isoformat ()
180+ return updated
148181 return None
149182
150183 @classmethod
0 commit comments