Coverage for filip/utils/cleanup.py: 89%
66 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-11-20 16:54 +0000
1"""
2Functions to clean up a tenant within a fiware based platform.
3"""
4import warnings
5from functools import wraps
7from pydantic import AnyHttpUrl, AnyUrl
8from requests import RequestException
9from typing import Callable, List, Union
10from filip.models import FiwareHeader
11from filip.clients.ngsi_v2 import \
12 ContextBrokerClient, \
13 IoTAClient, \
14 QuantumLeapClient
17def clear_context_broker(url: str=None,
18 fiware_header: FiwareHeader=None,
19 clear_registrations: bool = False,
20 cb_client: ContextBrokerClient = None
21 ):
22 """
23 Function deletes all entities, registrations and subscriptions for a
24 given fiware header. To use TLS connection you need to provide the cb_client parameter
25 as an argument with the Session object including the certificate and private key.
27 Note:
28 Always clear the devices first because the IoT-Agent will otherwise
29 through errors if it cannot find its registration anymore.
31 Args:
32 url: Url of the context broker service
33 fiware_header: header of the tenant
34 cb_client: enables TLS communication if created with Session object, only needed
35 for self-signed certificates
36 clear_registrations: Determines whether registrations should be deleted.
37 If registrations are deleted while devices with commands
38 still exist, these devices become unreachable.
39 Only set to true once such devices are cleared.
40 Returns:
41 None
42 """
43 assert url or cb_client, "Either url or client object must be given"
44 # create client
45 if cb_client is None:
46 client = ContextBrokerClient(url=url, fiware_header=fiware_header)
47 else:
48 client = cb_client
50 # clean entities
51 client.delete_entities(entities=client.get_entity_list())
53 # clear subscriptions
54 for sub in client.get_subscription_list():
55 client.delete_subscription(subscription_id=sub.id)
56 assert len(client.get_subscription_list()) == 0
58 # clear registrations
59 if clear_registrations:
60 for reg in client.get_registration_list():
61 client.delete_registration(registration_id=reg.id)
62 assert len(client.get_registration_list()) == 0
65def clear_iot_agent(url: Union[str, AnyHttpUrl] = None,
66 fiware_header: FiwareHeader = None,
67 iota_client: IoTAClient = None):
68 """
69 Function deletes all device groups and devices for a
70 given fiware header. To use TLS connection you need to provide the iota_client parameter
71 as an argument with the Session object including the certificate and private key.
73 Args:
74 url: Url of the iot agent service
75 fiware_header: header of the tenant
76 iota_client: enables TLS communication if created with Session object, only needed for self-signed certificates
78 Returns:
79 None
80 """
81 assert url or iota_client, "Either url or client object must be given"
82 # create client
83 if iota_client is None:
84 client = IoTAClient(url=url, fiware_header=fiware_header)
85 else:
86 client = iota_client
88 # clear registrations
89 for device in client.get_device_list():
90 client.delete_device(device_id=device.device_id)
91 assert len(client.get_device_list()) == 0
93 # clear groups
94 for group in client.get_group_list():
95 client.delete_group(resource=group.resource,
96 apikey=group.apikey)
97 assert len(client.get_group_list()) == 0
100def clear_quantumleap(url: str = None,
101 fiware_header: FiwareHeader = None,
102 ql_client: QuantumLeapClient = None):
103 """
104 Function deletes all data for a given fiware header. To use TLS connection you need to provide the ql_client parameter
105 as an argument with the Session object including the certificate and private key.
106 Args:
107 url: Url of the quantumleap service
108 fiware_header: header of the tenant
109 ql_client: enables TLS communication if created with Session object, only needed for self-signed certificates
111 Returns:
112 None
113 """
114 def handle_emtpy_db_exception(err: RequestException) -> None:
115 """
116 When the database is empty for request quantumleap returns a 404
117 error with a error message. This will be handled here
118 evaluating the empty database error as 'OK'
120 Args:
121 err: exception raised by delete function
122 """
123 if err.response.status_code == 404 \
124 and err.response.json().get('error', None) == 'Not Found':
125 pass
126 else:
127 raise
128 assert url or ql_client, "Either url or client object must be given"
129 # create client
130 if ql_client is None:
131 client = QuantumLeapClient(url=url, fiware_header=fiware_header)
132 else:
133 client = ql_client
135 # clear data
136 entities = []
137 try:
138 entities = client.get_entities()
139 except RequestException as err:
140 handle_emtpy_db_exception(err)
142 # will be executed for all found entities
143 for entity in entities:
144 client.delete_entity(entity_id=entity.entityId,
145 entity_type=entity.entityType)
148def clear_all(*,
149 fiware_header: FiwareHeader = None,
150 cb_url: str = None,
151 iota_url: Union[str, List[str]] = None,
152 ql_url: str = None,
153 cb_client: ContextBrokerClient = None,
154 iota_client: IoTAClient = None,
155 ql_client: QuantumLeapClient = None):
156 """
157 Clears all services that a url is provided for
159 Args:
160 fiware_header:
161 cb_url: url of the context broker service
162 iota_url: url of the IoT-Agent service
163 ql_url: url of the QuantumLeap service
164 cb_client: enables TLS communication if created with Session object, only needed
165 for self-signed certificates
166 iota_client: enables TLS communication if created with Session object, only needed
167 for self-signed certificates
168 ql_client: enables TLS communication if created with Session object, only needed
169 for self-signed certificates
171 Returns:
172 None
173 """
174 if iota_url is not None or iota_client is not None:
175 if iota_url is None:
176 # loop client
177 if isinstance(iota_client, IoTAClient):
178 iota_client = [iota_client]
179 for client in iota_client:
180 clear_iot_agent(fiware_header=fiware_header, iota_client=client)
181 else:
182 if isinstance(iota_url, (str, AnyUrl)):
183 iota_url = [iota_url]
184 for url in iota_url:
185 clear_iot_agent(url=url, fiware_header=fiware_header)
187 if cb_url is not None or cb_client is not None:
188 clear_context_broker(url=cb_url, fiware_header=fiware_header, cb_client=cb_client)
190 if ql_url is not None or ql_client is not None:
191 clear_quantumleap(url=ql_url, fiware_header=fiware_header, ql_client=ql_client)
194def clean_test(*,
195 fiware_service: str,
196 fiware_servicepath: str,
197 cb_url: str = None,
198 iota_url: Union[str, List[str]] = None,
199 ql_url: str = None,
200 cb_client: ContextBrokerClient = None,
201 iota_client: IoTAClient = None,
202 ql_client: QuantumLeapClient = None) -> Callable:
203 """
204 Decorator to clean up the server before and after the test
206 Note:
207 This does not substitute a proper TearDown method, because a failing
208 test will not execute the clean up after the error. Since this would
209 mean an unnecessary error handling. We actually want a test to fail
210 with proper messages.
212 Args:
213 fiware_service: tenant
214 fiware_servicepath: tenant path
215 cb_url: url of context broker service
216 iota_url: url of IoT-Agent service
217 ql_url: url of quantumleap service
218 cb_client: enables TLS communication if created with Session object, only needed for self-signed certificates
219 iota_client: enables TLS communication if created with Session object, only needed for self-signed certificates
220 ql_client: enables TLS communication if created with Session object, only needed for self-signed certificates
222 Returns:
223 Decorator for clean tests
224 """
225 fiware_header = FiwareHeader(service=fiware_service,
226 service_path=fiware_servicepath)
227 clear_all(fiware_header=fiware_header,
228 cb_url=cb_url,
229 iota_url=iota_url,
230 ql_url=ql_url,
231 cb_client=cb_client,
232 iota_client=iota_client,
233 ql_client=ql_client)
234 # Inner decorator function
235 def decorator(func):
236 # Wrapper function for the decorated function
237 @wraps(func)
238 def wrapper(*args, **kwargs):
239 return func(*args, **kwargs)
240 return wrapper
242 clear_all(fiware_header=fiware_header,
243 cb_url=cb_url,
244 iota_url=iota_url,
245 ql_url=ql_url,
246 cb_client=cb_client,
247 iota_client=iota_client,
248 ql_client=ql_client)
250 return decorator