From eb0760245ede5941f63cc612c53e2042f2666f90 Mon Sep 17 00:00:00 2001 From: Andrea Rosci Date: Thu, 11 Jun 2026 16:07:22 +0200 Subject: [PATCH] Add request batching for device and application operations Change-type: minor --- balena/models/application.py | 35 ++- balena/models/device.py | 285 +++++++++++--------- balena/request_batching.py | 138 ++++++++++ tests/functional/models/test_application.py | 26 ++ tests/functional/models/test_device.py | 115 +++++++- 5 files changed, 468 insertions(+), 131 deletions(-) create mode 100644 balena/request_batching.py diff --git a/balena/models/application.py b/balena/models/application.py index 54b7003a..2d2d086a 100644 --- a/balena/models/application.py +++ b/balena/models/application.py @@ -24,6 +24,7 @@ merge, with_supervisor_locked_error, ) +from ..request_batching import batch_resource_operation_factory from .device_type import DeviceType from .organization import Organization @@ -48,6 +49,12 @@ def __init__(self, pine: PineClient, settings: Settings, load_inner_models=True) self.membership = ApplicationMembership(pine, self) self.invite = ApplicationInvite(pine, self, settings) + self.__batch_application_operation = batch_resource_operation_factory( + get_all=self.get_all, + not_found_error=exceptions.ApplicationNotFound, + ambiguous_error=exceptions.AmbiguousApplication, + ) + def __get_access_filter(self): return { "is_directly_accessible_by__user": { @@ -481,13 +488,13 @@ def create( return self.__pine.post({"resource": "application", "body": body}) - # TODO: enable batch operations - def remove(self, slug_or_uuid_or_id: Union[str, int]) -> None: + def remove(self, slug_or_uuid_or_id_or_ids: Union[str, int, List[int]]) -> None: """ Remove application. Args: - slug_or_uuid_or_id (Union[str, int]): application slug (string), uuid (string) or id (number). + slug_or_uuid_or_id_or_ids (Union[str, int, List[int]]): application slug (string), uuid (string), + id (number) or ids (List[int]). Examples: >>> balena.models.application.remove('my_org/my_app') @@ -495,13 +502,29 @@ def remove(self, slug_or_uuid_or_id: Union[str, int]) -> None: >>> balena.models.application.remove(123) """ + if isinstance(slug_or_uuid_or_id_or_ids, list): + self.__batch_application_operation( + slug_or_uuid_or_id_or_ids, + parameter_name="slug_or_uuid_or_id_or_ids", + fn=lambda applications: self.__pine.delete( + { + "resource": "application", + "options": {"$filter": {"id": {"$in": [a["id"] for a in applications]}}}, + } + ), + ) + return + + if isinstance(slug_or_uuid_or_id_or_ids, str): + application_id = self.get_id(slug_or_uuid_or_id_or_ids) + else: + application_id = slug_or_uuid_or_id_or_ids try: - application_id = self.get_id(slug_or_uuid_or_id) self.__pine.delete({"resource": "application", "id": application_id}) except exceptions.RequestError as e: if e.status_code == 404: - raise exceptions.ApplicationNotFound(slug_or_uuid_or_id) - raise e + raise exceptions.ApplicationNotFound(str(slug_or_uuid_or_id_or_ids)) + raise def rename(self, slug_or_uuid_or_id: Union[str, int], new_name: str) -> None: """ diff --git a/balena/models/device.py b/balena/models/device.py index 88b77ce4..8e922383 100644 --- a/balena/models/device.py +++ b/balena/models/device.py @@ -32,6 +32,7 @@ merge, with_supervisor_locked_error, ) +from ..request_batching import batch_resource_operation_factory from .application import Application from .config import Config from .device_type import DeviceType @@ -114,6 +115,12 @@ def __init__(self, pine: PineClient, settings: Settings): self.service_var = DeviceServiceEnvVariable(pine, self, self.__application) self.history = DeviceHistory(pine, settings) + self.__batch_device_operation = batch_resource_operation_factory( + get_all=self.get_all, + not_found_error=exceptions.DeviceNotFound, + ambiguous_error=exceptions.AmbiguousDevice, + ) + self.__supervisor_address = os.environ.get("BALENA_SUPERVISOR_ADDRESS") self.__supervisor_api_key = os.environ.get("BALENA_SUPERVISOR_API_KEY") self.__on_device_app_id = os.environ.get("BALENA_APP_ID") @@ -162,38 +169,33 @@ def __set( body: Any, fn: Optional[Callable] = None, ) -> None: - if fn is None: - fn = self.__pine.patch - - if isinstance(uuid_or_id_or_ids, (int, str)): - if is_id(uuid_or_id_or_ids): - resource_id = uuid_or_id_or_ids - elif is_full_uuid(uuid_or_id_or_ids): - resource_id = {"uuid": uuid_or_id_or_ids} - else: - raise exceptions.InvalidParameter("uuid_or_id_or_ids", uuid_or_id_or_ids) - - fn( - { - "resource": "device", - "id": resource_id, - "body": body, - } - ) - else: - chunk_size = 200 - chunked_devices = [ - uuid_or_id_or_ids[i : i + chunk_size] # noqa: E203 type: ignore - for i in range(0, len(uuid_or_id_or_ids), chunk_size) - ] - for chunk in chunked_devices: - fn( - { - "resource": "device", - "options": {"$filter": {"id": {"$in": chunk}}}, - "body": body, - } + actual_fn = self.__pine.patch if fn is None else fn + + if isinstance(uuid_or_id_or_ids, list): + def _operation(devices: list) -> None: + device_filter = ( + {"id": devices[0]["id"]} if len(devices) == 1 + else {"id": {"$in": [d["id"] for d in devices]}} ) + query: Any = {"resource": "device", "options": {"$filter": device_filter}} + if body is not None: + query["body"] = body + actual_fn(query) + self.__batch_device_operation(uuid_or_id_or_ids, _operation) + return + + if uuid_or_id_or_ids == "": + raise exceptions.InvalidParameter("uuid_or_id_or_ids", uuid_or_id_or_ids) + resource_filter = ( + {"id": uuid_or_id_or_ids} if is_id(uuid_or_id_or_ids) else {"uuid": uuid_or_id_or_ids} + ) + devices = self.get_all({"$select": "id", "$filter": resource_filter}) + if not devices: + raise exceptions.DeviceNotFound(str(uuid_or_id_or_ids)) + query: Any = {"resource": "device", "id": devices[0]["id"]} + if body is not None: + query["body"] = body + actual_fn(query) def __check_local_mode_supported(self, device: TypeDevice): if not is_provisioned(device): @@ -667,56 +669,66 @@ def unset_custom_location(self, uuid_or_id_or_ids: Union[str, int, List[int]]) - self.set_custom_location(uuid_or_id_or_ids, {"latitude": "", "longitude": ""}) - # TODO: enable device batching def move( self, - uuid_or_id: Union[str, int], + uuid_or_id_or_ids: Union[str, int, List[int], List[str]], app_slug_or_uuid_or_id: Union[str, int], ): """ Move a device to another application. Args: - uuid_or_id (Union[str, int]): device uuid (str) or id (int). + uuid_or_id_or_ids (Union[str, int, List[int], List[str]]): device uuid (str) or id (int) + or array of uuids or ids. app_slug_or_uuid_or_id (Union[str, int]): application slug (string), uuid (string) or id (number). Examples: >>> balena.models.device.move(123, 'RPI1Test') """ - application_options = { - "$select": "id", - "$expand": { - "is_for__device_type": { - "$select": "is_of__cpu_architecture", - "$expand": {"is_of__cpu_architecture": {"$select": "slug"}}, - } + app = self.__application.get( + app_slug_or_uuid_or_id, + { + "$select": "id", + "$expand": { + "is_for__device_type": { + "$select": "is_of__cpu_architecture", + "$expand": {"is_of__cpu_architecture": {"$select": "slug"}}, + } + }, }, - } - - app = self.__application.get(app_slug_or_uuid_or_id, application_options) + ) app_cpu_arch_slug = app["is_for__device_type"][0]["is_of__cpu_architecture"][0]["slug"] - device_options = { - "$select": "is_of__device_type", - "$expand": { - "is_of__device_type": { - "$select": "is_of__cpu_architecture", - "$expand": { - "is_of__cpu_architecture": { - "$select": "slug", - } - }, + def _operation(devices: list, _owner_id: int) -> None: + for device in devices: + device_cpu_arch_slug = device["is_of__device_type"][0]["is_of__cpu_architecture"][0]["slug"] + if not self.__device_os.is_architecture_compatible_with(app_cpu_arch_slug, device_cpu_arch_slug): + raise exceptions.IncompatibleApplication(app_slug_or_uuid_or_id) + device_filter = ( + {"id": devices[0]["id"]} if len(devices) == 1 else {"id": {"$in": [d["id"] for d in devices]}} + ) + self.__pine.patch( + { + "resource": "device", + "options": {"$filter": device_filter}, + "body": {"belongs_to__application": app["id"]}, } - }, - } - - device = self.get(uuid_or_id, device_options) - device_cpu_arch_slug = device["is_of__device_type"][0]["is_of__cpu_architecture"][0]["slug"] - - if not self.__device_os.is_architecture_compatible_with(app_cpu_arch_slug, device_cpu_arch_slug): - raise exceptions.IncompatibleApplication(app_slug_or_uuid_or_id) + ) - self.__set(uuid_or_id, {"belongs_to__application": app["id"]}) + self.__batch_device_operation( + uuid_or_id_or_ids, + _operation, + options={ + "$select": "is_of__device_type", + "$expand": { + "is_of__device_type": { + "$select": "is_of__cpu_architecture", + "$expand": {"is_of__cpu_architecture": {"$select": "slug"}}, + } + }, + }, + group_by_navigation_property="belongs_to__application", + ) def __supervisor_request(self, method: str, path: str, body: Optional[AnyObject] = None): params = {"apikey": self.__supervisor_api_key} @@ -1622,10 +1634,9 @@ def is_tracking_application_release(self, uuid_or_id: Union[str, int]) -> bool: return not bool(self.get(uuid_or_id, {"$select": "is_pinned_on__release"})["is_pinned_on__release"]) - # TODO: enable device batching def pin_to_release( self, - uuid_or_id: Union[str, int], + uuid_or_id_or_ids: Union[str, int, List[int], List[str]], full_release_hash_or_id: Union[str, int], ) -> None: """ @@ -1633,42 +1644,50 @@ def pin_to_release( and not get updated when the current application release changes. Args: - uuid_or_id (Union[str, int]): device uuid (string) or id (int) + uuid_or_id_or_ids (Union[str, int, List[int], List[str]]): device uuid (string) or id (int) + or array of uuids or ids full_release_hash_or_id (Union[str, int]) : the hash of a successful release (string) or id (number) Examples: >>> balena.models.device.pin_to_release('49b2a', '45c90004de73557ded7274d4896a6db90ea61e36') """ + release_filter_prop = "id" if is_id(full_release_hash_or_id) else "commit" + release_cache = {} - device = self.get( - uuid_or_id, - { - "$select": "id", - "$expand": {"belongs_to__application": {"$select": "id"}}, - }, - ) - app_id = device["belongs_to__application"][0]["id"] - release_options = { - "$top": 1, - "$select": "id", - "$filter": { - "status": "success", - "belongs_to__application": app_id, - }, - "$orderby": "created_at desc", - } - if is_id(full_release_hash_or_id): - release_options["$filter"]["id"] = full_release_hash_or_id - else: - release_options["$filter"]["commit"] = full_release_hash_or_id + def _get_release(app_id: int): + if app_id not in release_cache: + release_cache[app_id] = self.__release.get( + full_release_hash_or_id, + { + "$top": 1, + "$select": "id", + "$filter": { + release_filter_prop: full_release_hash_or_id, + "status": "success", + "belongs_to__application": app_id, + }, + "$orderby": "created_at desc", + }, + ) + return release_cache[app_id] - release = self.__release.get(full_release_hash_or_id, release_options) - self.__pine.patch( - { - "resource": "device", - "id": device["id"], - "body": {"is_pinned_on__release": release["id"]}, - } + def _operation(devices: list, app_id: int) -> None: + release = _get_release(app_id) + device_filter = ( + {"id": devices[0]["id"]} if len(devices) == 1 else {"id": {"$in": [d["id"] for d in devices]}} + ) + self.__pine.patch( + { + "resource": "device", + "options": {"$filter": device_filter}, + "body": {"is_pinned_on__release": release["id"]}, + } + ) + + self.__batch_device_operation( + uuid_or_id_or_ids, + _operation, + group_by_navigation_property="belongs_to__application", ) def track_application_release(self, uuid_or_id_or_ids: Union[str, int, List[int]]) -> None: @@ -1681,49 +1700,69 @@ def track_application_release(self, uuid_or_id_or_ids: Union[str, int, List[int] self.__set(uuid_or_id_or_ids, {"is_pinned_on__release": None}) - # TODO: enable device batching def pin_to_supervisor_release( self, - uuid_or_id: Union[str, int], + uuid_or_id_or_ids: Union[str, int, List[int], List[str]], supervisor_version_or_id: Union[str, int], ) -> None: """ Set a specific device to run a particular supervisor release. Args: - uuid_or_id (Union[str, int]): device uuid (string) or id (int) + uuid_or_id_or_ids (Union[str, int, List[int], List[str]]): device uuid (string) or id (int) + or array of uuids or ids supervisor_version_or_id (Union[str, int]): the version of a released supervisor (string) or id (number) Examples: >>> balena.models.device.pin_to_supervisor_release('f55dcdd9ada04b11b4d05c1f1c3b4e72', 'v13.0.0') """ - device = self.get( - uuid_or_id, - { - "$select": ["id", "supervisor_version", "os_version"], + release_filter_prop = "id" if is_id(supervisor_version_or_id) else "raw_version" + release_cache = {} + + def _get_release(cpu_arch_id: int): + if cpu_arch_id not in release_cache: + try: + release = self.__device_os.get_supervisor_releases_for_cpu_architecture( + cpu_arch_id, + { + "$top": 1, + "$select": "id", + "$filter": {release_filter_prop: supervisor_version_or_id}, + }, + )[0] + except IndexError: + raise Exception(f"Supervisor release not found {supervisor_version_or_id}") + release_cache[cpu_arch_id] = release + return release_cache[cpu_arch_id] + + def _operation(devices: list) -> None: + for device in devices: + ensure_version_compatibility(device["supervisor_version"], MIN_SUPERVISOR_MC_API, "supervisor") + ensure_version_compatibility(device["os_version"], MIN_OS_MC, "host OS") + + devices_by_arch = {} + for device in devices: + cpu_arch_id = device["is_of__device_type"][0]["is_of__cpu_architecture"]["__id"] + devices_by_arch.setdefault(cpu_arch_id, []).append(device) + + for cpu_arch_id, arch_devices in devices_by_arch.items(): + release = _get_release(cpu_arch_id) + ids = [d["id"] for d in arch_devices] + arch_filter = {"id": arch_devices[0]["id"]} if len(arch_devices) == 1 else {"id": {"$in": ids}} + self.__pine.patch( + { + "resource": "device", + "options": {"$filter": arch_filter}, + "body": {"should_be_managed_by__release": release["id"]}, + } + ) + + self.__batch_device_operation( + uuid_or_id_or_ids, + _operation, + options={ + "$select": ["supervisor_version", "os_version"], "$expand": {"is_of__device_type": {"$select": "is_of__cpu_architecture"}}, }, ) - cpu_arch_id = device["is_of__device_type"][0]["is_of__cpu_architecture"]["__id"] - - release_options = { - "$top": 1, - "$select": "id", - "$filter": {"id" if is_id(supervisor_version_or_id) else "raw_version": supervisor_version_or_id}, - } - - try: - release = self.__device_os.get_supervisor_releases_for_cpu_architecture(cpu_arch_id, release_options)[0] - except IndexError: - raise Exception(f"Supervisor release not found {supervisor_version_or_id}") - - ensure_version_compatibility(device["supervisor_version"], MIN_SUPERVISOR_MC_API, "supervisor") - ensure_version_compatibility(device["os_version"], MIN_OS_MC, "host OS") - self.__pine.patch( - { - "resource": "device", - "id": device["id"], - "body": {"should_be_managed_by__release": release["id"]}, - } - ) def start_os_update( self, diff --git a/balena/request_batching.py b/balena/request_batching.py new file mode 100644 index 00000000..dbe694af --- /dev/null +++ b/balena/request_batching.py @@ -0,0 +1,138 @@ +from collections import defaultdict +from typing import Any, Callable, Dict, List, Optional, Type, Union + +from .utils import merge + +NUMERIC_ID_CHUNK_SIZE = 200 +STRING_ID_CHUNK_SIZE = 50 + + +def _chunk(lst: list, size: int) -> list: + return [lst[i : i + size] for i in range(0, len(lst), size)] # noqa: E203 + + +def _group_by_map(items: list, key_fn: Callable) -> Dict: + result: Dict[Any, list] = defaultdict(list) + for item in items: + result[key_fn(item)].append(item) + return dict(result) + + +def batch_resource_operation_factory( + get_all: Callable, + not_found_error: Type[Exception], + ambiguous_error: Type[Exception], + chunk_size: Optional[Union[int, Dict]] = None, +) -> Callable: + """ + Factory that creates a batch resource operation function. + + Args: + get_all (Callable): function to retrieve all resources matching given options. + not_found_error (Type[Exception]): exception raised when a resource is not found. + ambiguous_error (Type[Exception]): exception raised when a UUID matches multiple resources. + chunk_size (Optional[Union[int, Dict]]): optional overrides for numeric_id and string_id chunk sizes. + + Returns: + Callable: batch operation function. + """ + + if isinstance(chunk_size, int): + numeric_chunk = chunk_size + string_chunk = chunk_size + else: + numeric_chunk = (chunk_size or {}).get("numeric_id", NUMERIC_ID_CHUNK_SIZE) + string_chunk = (chunk_size or {}).get("string_id", STRING_ID_CHUNK_SIZE) + + def batch_resource_operation( + uuid_or_id_or_array: Union[str, int, List[int], List[str]], + fn: Callable, + parameter_name: str = "uuid_or_id_or_array", + options: Optional[Any] = None, + group_by_navigation_property: Optional[str] = None, + ) -> None: + """ + Perform a batch operation on resources identified by uuid(s) or id(s). + + Args: + uuid_or_id_or_array (Union[str, int, List[int], List[str]]): resource identifier(s). + fn (Callable): function called with resolved items, and owner id when group_by_navigation_property is set. + parameter_name (str): parameter name used in error messages. + options (Optional[AnyObject]): extra pine options to merge. + group_by_navigation_property (Optional[str]): navigation property to group items by before calling fn. + """ + from .exceptions import InvalidParameter + + if uuid_or_id_or_array == "": + raise InvalidParameter(parameter_name, uuid_or_id_or_array) + + if isinstance(uuid_or_id_or_array, list): + if not uuid_or_id_or_array: + raise InvalidParameter(parameter_name, uuid_or_id_or_array) + first_type = type(uuid_or_id_or_array[0]) + for param in uuid_or_id_or_array: + if not isinstance(param, (int, str)): + raise InvalidParameter("uuid_or_id_or_array", uuid_or_id_or_array) + if type(param) is not first_type: + raise InvalidParameter("uuid_or_id_or_array", uuid_or_id_or_array) + if isinstance(param, str) and len(param) not in (32, 62): + raise InvalidParameter("uuid_or_id_or_array", uuid_or_id_or_array) + + if isinstance(uuid_or_id_or_array, list): + chunks = ( + _chunk(uuid_or_id_or_array, string_chunk) + if isinstance(uuid_or_id_or_array[0], str) + else _chunk(uuid_or_id_or_array, numeric_chunk) + ) + else: + chunks = [uuid_or_id_or_array] + + items = [] + for chunk in chunks: + if isinstance(chunk, list): + resource_filter = {"uuid": {"$in": chunk}} if isinstance(chunk[0], str) else {"id": {"$in": chunk}} + else: + resource_filter = {"uuid": chunk} if isinstance(chunk, str) else {"id": chunk} + + select: List[str] = ["id"] + if isinstance(chunk, list) and isinstance(chunk[0], str): + select.append("uuid") + if group_by_navigation_property: + select.append(group_by_navigation_property) + + options_without_select = options + if options and "$select" in options: + opt_select = options["$select"] + if opt_select != "*": + extra = opt_select if isinstance(opt_select, list) else [opt_select] + select = list(set(select + extra)) + options_without_select = {k: v for k, v in options.items() if k != "$select"} + + items.extend(get_all(merge({"$select": select, "$filter": resource_filter}, options_without_select))) + + if not items: + raise not_found_error(str(uuid_or_id_or_array)) + + if isinstance(uuid_or_id_or_array, str) and len(items) > 1: + raise ambiguous_error(uuid_or_id_or_array) + + if isinstance(uuid_or_id_or_array, list): + identifier_key = "uuid" if isinstance(uuid_or_id_or_array[0], str) else "id" + found = {item[identifier_key] for item in items} + for identifier in uuid_or_id_or_array: + if identifier not in found: + raise not_found_error(identifier) + + if group_by_navigation_property: + groups = _group_by_map(items, lambda item: item[group_by_navigation_property]["__id"]) + else: + groups = {None: items} + + for owner_id, group_items in groups.items(): + for chunked_items in _chunk(group_items, numeric_chunk): + if group_by_navigation_property: + fn(chunked_items, owner_id) + else: + fn(chunked_items) + + return batch_resource_operation diff --git a/tests/functional/models/test_application.py b/tests/functional/models/test_application.py index 7514f04d..7f8078c6 100644 --- a/tests/functional/models/test_application.py +++ b/tests/functional/models/test_application.py @@ -260,6 +260,32 @@ def test_34_membership_remove(self): membership_list = self.balena.models.application.membership.get_all() self.assertEqual(0, len(membership_list)) + def test_35_remove_not_found(self): + with self.assertRaises(self.helper.balena_exceptions.ApplicationNotFound) as cm: + self.balena.models.application.remove(999999) + self.assertIn("Application not found: 999999", cm.exception.message) + + with self.assertRaises(self.helper.balena_exceptions.ApplicationNotFound): + self.balena.models.application.remove(f"{self.org_handle}/nonexistentapp") + + def test_36_remove_batch(self): + app1 = self.balena.models.application.create("BatchApp1", "raspberry-pi2", self.org_id) + app2 = self.balena.models.application.create("BatchApp2", "raspberry-pi2", self.org_id) + + self.balena.models.application.remove([app1["id"], app2["id"]]) + + with self.assertRaises(self.helper.balena_exceptions.ApplicationNotFound): + self.balena.models.application.get(app1["id"]) + with self.assertRaises(self.helper.balena_exceptions.ApplicationNotFound): + self.balena.models.application.get(app2["id"]) + + app3 = self.balena.models.application.create("BatchApp3", "raspberry-pi2", self.org_id) + with self.assertRaises(self.helper.balena_exceptions.ApplicationNotFound): + self.balena.models.application.remove([app3["id"], 999999]) + + self.assertIsNotNone(self.balena.models.application.get(app3["id"])) + self.balena.models.application.remove(app3["id"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/functional/models/test_device.py b/tests/functional/models/test_device.py index 9e39a06d..dc5931a3 100644 --- a/tests/functional/models/test_device.py +++ b/tests/functional/models/test_device.py @@ -394,10 +394,10 @@ def test_30_remove(self): with self.assertRaises(self.helper.balena_exceptions.InvalidParameter): self.balena.models.device.remove("") - with self.assertRaises(self.helper.balena_exceptions.InvalidParameter): + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): self.balena.models.device.remove("abc") - with self.assertRaises(self.helper.balena_exceptions.InvalidParameter): + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): self.balena.models.device.remove(uuid[0:10]) device_uuids = [device["uuid"] for device in self.balena.models.device.get_all()] @@ -411,6 +411,117 @@ def test_30_remove(self): self.assertNotIn(uuid, device_uuids) self.assertNotIn(uuid2, device_uuids) + def test_31_remove_not_found(self): + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound) as cm: + self.balena.models.device.remove("a2df0000025c4223b4efe2b66f3e370a") + self.assertIn("Device not found: a2df0000025c4223b4efe2b66f3e370a", cm.exception.message) + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound) as cm: + self.balena.models.device.remove(999999) + self.assertIn("Device not found: 999999", cm.exception.message) + + def test_32_remove_by_id(self): + uuid = self.balena.models.device.generate_uuid() + device = self.balena.models.device.register(self.app["id"], uuid) + self.balena.models.device.remove(device["id"]) + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.get(device["uuid"]) + + def test_33_remove_batch(self): + uuid1 = self.balena.models.device.generate_uuid() + uuid2 = self.balena.models.device.generate_uuid() + device1 = self.balena.models.device.register(self.app["id"], uuid1) + device2 = self.balena.models.device.register(self.app["id"], uuid2) + + self.balena.models.device.remove([device1["id"], device2["id"]]) + + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.get(device1["uuid"]) + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.get(device2["uuid"]) + + uuid3 = self.balena.models.device.generate_uuid() + device3 = self.balena.models.device.register(self.app["id"], uuid3) + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.remove([device3["id"], 999999]) + + self.assertIsNotNone(self.balena.models.device.get(device3["uuid"])) + self.balena.models.device.remove(device3["id"]) + + def test_34_pin_to_release_batch(self): + app_info = self.helper.create_multicontainer_app(app_name="FooBatchPin") + device2 = self.balena.models.device.register( + app_info["app"]["id"], self.balena.models.device.generate_uuid() + ) + device1_id = app_info["device"]["id"] + device2_id = device2["id"] + + self.balena.models.device.pin_to_release( + [device1_id, device2_id], app_info["old_release"]["commit"] + ) + self.assertFalse( + self.balena.models.device.is_tracking_application_release(app_info["device"]["uuid"]) + ) + self.assertFalse( + self.balena.models.device.is_tracking_application_release(device2["uuid"]) + ) + + self.balena.models.device.track_application_release([device1_id, device2_id]) + self.assertTrue( + self.balena.models.device.is_tracking_application_release(app_info["device"]["uuid"]) + ) + self.assertTrue( + self.balena.models.device.is_tracking_application_release(device2["uuid"]) + ) + + self.balena.models.device.pin_to_release( + [app_info["device"]["uuid"], device2["uuid"]], app_info["old_release"]["id"] + ) + self.assertFalse( + self.balena.models.device.is_tracking_application_release(app_info["device"]["uuid"]) + ) + self.assertFalse( + self.balena.models.device.is_tracking_application_release(device2["uuid"]) + ) + + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.pin_to_release( + [device1_id, 999999], app_info["current_release"]["commit"] + ) + self.assertFalse( + self.balena.models.device.is_tracking_application_release(app_info["device"]["uuid"]) + ) + + def test_35_pin_to_supervisor_release_batch(self): + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.pin_to_supervisor_release(999999, "v13.0.0") + + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.pin_to_supervisor_release([999998, 999999], "v13.0.0") + + def test_36_move_batch(self): + app_src = self.balena.models.application.create( + "FooBatchMoveSrc", "raspberry-pi2", self.helper.default_organization["id"] + ) + app_dst = self.balena.models.application.create( + "FooBatchMoveDst", "raspberry-pi2", self.helper.default_organization["id"] + ) + app_incompatible = self.balena.models.application.create( + "FooBatchMoveIncompat", "intel-nuc", self.helper.default_organization["id"] + ) + + device1 = self.balena.models.device.register(app_src["id"], self.balena.models.device.generate_uuid()) + device2 = self.balena.models.device.register(app_src["id"], self.balena.models.device.generate_uuid()) + + self.balena.models.device.move([device1["id"], device2["id"]], app_dst["slug"]) + self.assertEqual(self.balena.models.device.get_application_name(device1["uuid"]), app_dst["app_name"]) + self.assertEqual(self.balena.models.device.get_application_name(device2["uuid"]), app_dst["app_name"]) + + with self.assertRaises(self.helper.balena_exceptions.IncompatibleApplication): + self.balena.models.device.move([device1["id"], device2["id"]], app_incompatible["slug"]) + + with self.assertRaises(self.helper.balena_exceptions.DeviceNotFound): + self.balena.models.device.move([device1["id"], 999999], app_dst["slug"]) + if __name__ == "__main__": unittest.main()