Add unit test for Coriolis api v1 views

coriolis/api/v1/diagnostics.py

@@ -23,7 +23,7 @@ class DiagnosticsController(api_wsgi.Controller):
             diagnostics.get_diagnostics_policy_label("get"))
 
         return diagnostic_view.collection(
-            req, self._diag_api.get(context))
+            self._diag_api.get(context))
 
 
 def create_resource():

coriolis/api/v1/endpoint_destination_minion_pool_options.py

@@ -37,7 +37,6 @@ class EndpointDestinationMinionPoolOptionsController(api_wsgi.Controller):
 
         return (endpoint_options_view.
                 destination_minion_pool_options_collection)(
-            req,
             (self._minion_pool_options_api.
              get_endpoint_destination_minion_pool_options)(
                 context, endpoint_id, env=env, option_names=options))

coriolis/api/v1/endpoint_destination_options.py

@@ -36,7 +36,6 @@ class EndpointDestinationOptionsController(api_wsgi.Controller):
             options = {}
 
         return endpoint_options_view.destination_options_collection(
-            req,
             self._destination_options_api.get_endpoint_destination_options(
                 context, endpoint_id, env=env, option_names=options))
 

coriolis/api/v1/endpoint_instances.py

@@ -32,7 +32,7 @@ class EndpointInstanceController(api_wsgi.Controller):
             env = {}
 
         return endpoint_resources_view.instances_collection(
-            req, self._instance_api.get_endpoint_instances(
+            self._instance_api.get_endpoint_instances(
                 context, endpoint_id, env, marker, limit,
                 instance_name_pattern))
 
@@ -52,7 +52,7 @@ class EndpointInstanceController(api_wsgi.Controller):
             env = {}
 
         return endpoint_resources_view.instance_single(
-            req, self._instance_api.get_endpoint_instance(
+            self._instance_api.get_endpoint_instance(
                 req.environ['coriolis.context'], endpoint_id, env, id))
 
 

coriolis/api/v1/endpoint_networks.py

@@ -28,7 +28,7 @@ class EndpointNetworkController(api_wsgi.Controller):
             env = {}
 
         return endpoint_resources_view.networks_collection(
-            req, self._network_api.get_endpoint_networks(
+            self._network_api.get_endpoint_networks(
                 context, endpoint_id, env))
 
 

coriolis/api/v1/endpoint_source_minion_pool_options.py

@@ -36,7 +36,6 @@ class EndpointSourceMinionPoolOptionsController(api_wsgi.Controller):
             options = {}
 
         return endpoint_options_view.source_minion_pool_options_collection(
-            req,
             (self._minion_pool_options_api.
              get_endpoint_source_minion_pool_options)(
                 context, endpoint_id, env=env, option_names=options))

coriolis/api/v1/endpoint_source_options.py

@@ -36,7 +36,6 @@ class EndpointSourceOptionsController(api_wsgi.Controller):
             options = {}
 
         return endpoint_options_view.source_options_collection(
-            req,
             self._source_options_api.get_endpoint_source_options(
                 context, endpoint_id, env=env, option_names=options))
 

coriolis/api/v1/endpoint_storage.py

@@ -28,7 +28,7 @@ class EndpointStorageController(api_wsgi.Controller):
             env = {}
 
         return endpoint_resources_view.storage_collection(
-            req, self._storage_api.get_endpoint_storage(
+            self._storage_api.get_endpoint_storage(
                 context, endpoint_id, env))
 
 

coriolis/api/v1/endpoints.py

@@ -27,13 +27,13 @@ class EndpointController(api_wsgi.Controller):
         if not endpoint:
             raise exc.HTTPNotFound()
 
-        return endpoint_view.single(req, endpoint)
+        return endpoint_view.single(endpoint)
 
     def index(self, req):
         context = req.environ["coriolis.context"]
         context.can(endpoint_policies.get_endpoints_policy_label("list"))
         return endpoint_view.collection(
-            req, self._endpoint_api.get_endpoints(context))
+            self._endpoint_api.get_endpoints(context))
 
     @api_utils.format_keyerror_message(resource='endpoint', method='create')
     def _validate_create_body(self, body):
@@ -52,7 +52,7 @@ class EndpointController(api_wsgi.Controller):
         context.can(endpoint_policies.get_endpoints_policy_label("create"))
         (name, endpoint_type, description,
          connection_info, mapped_regions) = self._validate_create_body(body)
-        return endpoint_view.single(req, self._endpoint_api.create(
+        return endpoint_view.single(self._endpoint_api.create(
             context, name, endpoint_type, description, connection_info,
             mapped_regions))
 
@@ -69,7 +69,7 @@ class EndpointController(api_wsgi.Controller):
         context = req.environ["coriolis.context"]
         context.can(endpoint_policies.get_endpoints_policy_label("update"))
         updated_values = self._validate_update_body(body)
-        return endpoint_view.single(req, self._endpoint_api.update(
+        return endpoint_view.single(self._endpoint_api.update(
             req.environ['coriolis.context'], id, updated_values))
 
     def delete(self, req, id):

coriolis/api/v1/migrations.py

@@ -41,7 +41,7 @@ class MigrationController(api_wsgi.Controller):
         if not migration:
             raise exc.HTTPNotFound()
 
-        return migration_view.single(req, migration)
+        return migration_view.single(migration)
 
     def _list(self, req):
         show_deleted = api_utils._get_show_deleted(
@@ -50,7 +50,7 @@ class MigrationController(api_wsgi.Controller):
         context.show_deleted = show_deleted
         context.can(migration_policies.get_migrations_policy_label("list"))
         return migration_view.collection(
-            req, self._migration_api.get_migrations(
+            self._migration_api.get_migrations(
                 context,
                 include_tasks=CONF.api.include_task_info_in_migrations_api,
                 include_task_info=CONF.api.include_task_info_in_migrations_api
@@ -173,7 +173,7 @@ class MigrationController(api_wsgi.Controller):
                 skip_os_morphing=skip_os_morphing,
                 user_scripts=user_scripts)
 
-        return migration_view.single(req, migration)
+        return migration_view.single(migration)
 
     def delete(self, req, id):
         context = req.environ['coriolis.context']

coriolis/api/v1/minion_pool_actions.py

@@ -23,7 +23,7 @@ class MinionPoolActionsController(api_wsgi.Controller):
                 "allocate"))
         try:
             return minion_pool_view.single(
-                req, self.minion_pool_api.allocate_minion_pool(
+                self.minion_pool_api.allocate_minion_pool(
                     context, id))
         except exception.NotFound as ex:
             raise exc.HTTPNotFound(explanation=ex.msg)
@@ -38,7 +38,7 @@ class MinionPoolActionsController(api_wsgi.Controller):
                 "refresh"))
         try:
             return minion_pool_view.single(
-                req, self.minion_pool_api.refresh_minion_pool(
+                self.minion_pool_api.refresh_minion_pool(
                     context, id))
         except exception.NotFound as ex:
             raise exc.HTTPNotFound(explanation=ex.msg)
@@ -54,7 +54,7 @@ class MinionPoolActionsController(api_wsgi.Controller):
         force = (body["deallocate"] or {}).get("force", False)
         try:
             return minion_pool_view.single(
-                req, self.minion_pool_api.deallocate_minion_pool(
+                self.minion_pool_api.deallocate_minion_pool(
                     context, id, force=force))
         except exception.NotFound as ex:
             raise exc.HTTPNotFound(explanation=ex.msg)

coriolis/api/v1/minion_pools.py

@@ -29,13 +29,13 @@ class MinionPoolController(api_wsgi.Controller):
         if not minion_pool:
             raise exc.HTTPNotFound()
 
-        return minion_pool_view.single(req, minion_pool)
+        return minion_pool_view.single(minion_pool)
 
     def index(self, req):
         context = req.environ["coriolis.context"]
         context.can(pools_policies.get_minion_pools_policy_label("list"))
         return minion_pool_view.collection(
-            req, self._minion_pool_api.get_minion_pools(context))
+            self._minion_pool_api.get_minion_pools(context))
 
     def _check_pool_retention_strategy(self, pool_retention_strategy):
         if not pool_retention_strategy:
@@ -137,7 +137,7 @@ class MinionPoolController(api_wsgi.Controller):
          minimum_minions, maximum_minions, minion_max_idle_time,
          minion_retention_strategy, notes, skip_allocation) = (
             self._validate_create_body(context, body))
-        return minion_pool_view.single(req, self._minion_pool_api.create(
+        return minion_pool_view.single(self._minion_pool_api.create(
             context, name, endpoint_id, pool_platform, pool_os_type,
             environment_options, minimum_minions, maximum_minions,
             minion_max_idle_time, minion_retention_strategy, notes=notes,
@@ -197,7 +197,7 @@ class MinionPoolController(api_wsgi.Controller):
         context.can(pools_policies.get_minion_pools_policy_label("update"))
         updated_values = self._validate_update_body(id, context, body)
         return minion_pool_view.single(
-            req, self._minion_pool_api.update(
+            self._minion_pool_api.update(
                 req.environ['coriolis.context'], id, updated_values))
 
     def delete(self, req, id):

coriolis/api/v1/regions.py

@@ -26,13 +26,13 @@ class RegionController(api_wsgi.Controller):
         if not region:
             raise exc.HTTPNotFound()
 
-        return region_view.single(req, region)
+        return region_view.single(region)
 
     def index(self, req):
         context = req.environ["coriolis.context"]
         context.can(region_policies.get_regions_policy_label("list"))
         return region_view.collection(
-            req, self._region_api.get_regions(context))
+            self._region_api.get_regions(context))
 
     @api_utils.format_keyerror_message(resource='region', method='create')
     def _validate_create_body(self, body):
@@ -46,7 +46,7 @@ class RegionController(api_wsgi.Controller):
         context = req.environ["coriolis.context"]
         context.can(region_policies.get_regions_policy_label("create"))
         (name, description, enabled) = self._validate_create_body(body)
-        return region_view.single(req, self._region_api.create(
+        return region_view.single(self._region_api.create(
             context, region_name=name, description=description,
             enabled=enabled))
 
@@ -60,7 +60,7 @@ class RegionController(api_wsgi.Controller):
         context = req.environ["coriolis.context"]
         context.can(region_policies.get_regions_policy_label("update"))
         updated_values = self._validate_update_body(body)
-        return region_view.single(req, self._region_api.update(
+        return region_view.single(self._region_api.update(
             req.environ['coriolis.context'], id, updated_values))
 
     def delete(self, req, id):

coriolis/api/v1/replica_actions.py

@@ -22,7 +22,7 @@ class ReplicaActionsController(api_wsgi.Controller):
             replica_policies.get_replicas_policy_label("delete_disks"))
         try:
             return replica_tasks_execution_view.single(
-                req, self._replica_api.delete_disks(context, id))
+                self._replica_api.delete_disks(context, id))
         except exception.NotFound as ex:
             raise exc.HTTPNotFound(explanation=ex.msg)
         except exception.InvalidParameterValue as ex:

coriolis/api/v1/replica_schedules.py

@@ -31,7 +31,7 @@ class ReplicaScheduleController(api_wsgi.Controller):
         if not schedule:
             raise exc.HTTPNotFound()
 
-        return replica_schedule_view.single(req, schedule)
+        return replica_schedule_view.single(schedule)
 
     def index(self, req, replica_id):
         context = req.environ["coriolis.context"]
@@ -41,7 +41,7 @@ class ReplicaScheduleController(api_wsgi.Controller):
         show_expired = strutils.bool_from_string(
             req.GET.get("show_expired", True), strict=True)
         return replica_schedule_view.collection(
-            req, self._schedule_api.get_schedules(
+            self._schedule_api.get_schedules(
                 context, replica_id, expired=show_expired))
 
     def _validate_schedule(self, schedule):
@@ -112,7 +112,7 @@ class ReplicaScheduleController(api_wsgi.Controller):
         except Exception as err:
             raise exception.InvalidInput(err)
 
-        return replica_schedule_view.single(req, self._schedule_api.create(
+        return replica_schedule_view.single(self._schedule_api.create(
             context, replica_id, schedule, enabled, exp_date, shutdown))
 
     def update(self, req, replica_id, id, body):
@@ -128,7 +128,7 @@ class ReplicaScheduleController(api_wsgi.Controller):
         except Exception as err:
             raise exception.InvalidInput(err)
 
-        return replica_schedule_view.single(req, self._schedule_api.update(
+        return replica_schedule_view.single(self._schedule_api.update(
             context, replica_id, id, update_values))
 
     def delete(self, req, replica_id, id):

coriolis/api/v1/replica_tasks_executions.py

@@ -24,7 +24,7 @@ class ReplicaTasksExecutionController(api_wsgi.Controller):
         if not execution:
             raise exc.HTTPNotFound()
 
-        return replica_tasks_execution_view.single(req, execution)
+        return replica_tasks_execution_view.single(execution)
 
     def index(self, req, replica_id):
         context = req.environ["coriolis.context"]
@@ -32,7 +32,7 @@ class ReplicaTasksExecutionController(api_wsgi.Controller):
             executions_policies.get_replica_executions_policy_label("list"))
 
         return replica_tasks_execution_view.collection(
-            req, self._replica_tasks_execution_api.get_executions(
+            self._replica_tasks_execution_api.get_executions(
                 context, replica_id, include_tasks=False))
 
     def detail(self, req, replica_id):
@@ -41,7 +41,7 @@ class ReplicaTasksExecutionController(api_wsgi.Controller):
             executions_policies.get_replica_executions_policy_label("show"))
 
         return replica_tasks_execution_view.collection(
-            req, self._replica_tasks_execution_api.get_executions(
+            self._replica_tasks_execution_api.get_executions(
                 context, replica_id, include_tasks=True))
 
     def create(self, req, replica_id, body):
@@ -55,7 +55,7 @@ class ReplicaTasksExecutionController(api_wsgi.Controller):
         shutdown_instances = execution_body.get("shutdown_instances", False)
 
         return replica_tasks_execution_view.single(
-            req, self._replica_tasks_execution_api.create(
+            self._replica_tasks_execution_api.create(
                 context, replica_id, shutdown_instances))
 
     def delete(self, req, replica_id, id):

coriolis/api/v1/replicas.py

@@ -41,7 +41,7 @@ class ReplicaController(api_wsgi.Controller):
         if not replica:
             raise exc.HTTPNotFound()
 
-        return replica_view.single(req, replica)
+        return replica_view.single(replica)
 
     def _list(self, req):
         show_deleted = api_utils._get_show_deleted(
@@ -51,7 +51,7 @@ class ReplicaController(api_wsgi.Controller):
         context.can(replica_policies.get_replicas_policy_label("list"))
         include_task_info = CONF.api.include_task_info_in_replicas_api
         return replica_view.collection(
-            req, self._replica_api.get_replicas(
+            self._replica_api.get_replicas(
                 context,
                 include_tasks_executions=include_task_info,
                 include_task_info=include_task_info))
@@ -135,7 +135,7 @@ class ReplicaController(api_wsgi.Controller):
          instance_osmorphing_minion_pool_mappings, user_scripts) = (
             self._validate_create_body(context, body))
 
-        return replica_view.single(req, self._replica_api.create(
+        return replica_view.single(self._replica_api.create(
             context, origin_endpoint_id, destination_endpoint_id,
             origin_minion_pool_id, destination_minion_pool_id,
             instance_osmorphing_minion_pool_mappings, source_environment,
@@ -342,8 +342,8 @@ class ReplicaController(api_wsgi.Controller):
         updated_values = self._validate_update_body(id, context, body)
         try:
             return replica_tasks_execution_view.single(
-                req, self._replica_api.update(req.environ['coriolis.context'],
-                                              id, updated_values))
+                self._replica_api.update(req.environ['coriolis.context'],
+                                         id, updated_values))
         except exception.NotFound as ex:
             raise exc.HTTPNotFound(explanation=ex.msg)
         except exception.InvalidParameterValue as ex:

coriolis/api/v1/services.py

@@ -26,13 +26,13 @@ class ServiceController(api_wsgi.Controller):
         if not service:
             raise exc.HTTPNotFound()
 
-        return service_view.single(req, service)
+        return service_view.single(service)
 
     def index(self, req):
         context = req.environ["coriolis.context"]
         context.can(service_policies.get_services_policy_label("list"))
         return service_view.collection(
-            req, self._service_api.get_services(context))
+            self._service_api.get_services(context))
 
     @api_utils.format_keyerror_message(resource='service', method='create')
     def _validate_create_body(self, body):
@@ -49,7 +49,7 @@ class ServiceController(api_wsgi.Controller):
         context.can(service_policies.get_services_policy_label("create"))
         (host, binary, topic, mapped_regions, enabled) = (
             self._validate_create_body(body))
-        return service_view.single(req, self._service_api.create(
+        return service_view.single(self._service_api.create(
             context, host=host, binary=binary, topic=topic,
             mapped_regions=mapped_regions, enabled=enabled))
 
@@ -63,7 +63,7 @@ class ServiceController(api_wsgi.Controller):
         context = req.environ["coriolis.context"]
         context.can(service_policies.get_services_policy_label("update"))
         updated_values = self._validate_update_body(body)
-        return service_view.single(req, self._service_api.update(
+        return service_view.single(self._service_api.update(
             req.environ['coriolis.context'], id, updated_values))
 
     def delete(self, req, id):

coriolis/api/v1/views/diagnostic_view.py

@@ -2,9 +2,9 @@
 # All Rights Reserved.
 
 
-def single(req, diag):
+def single(diag):
     return {"diagnostic": diag}
 
 
-def collection(req, diag):
+def collection(diag):
     return {'diagnostics': diag}

coriolis/api/v1/views/endpoint_options_view.py

@@ -1,38 +1,29 @@
 # Copyright 2020 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_opt(req, option, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
-    return dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in option.items()))
-
-
-def destination_minion_pool_options_collection(req, destination_pool_options):
+def destination_minion_pool_options_collection(destination_pool_options,
+                                               keys=None):
     formatted_opts = [
-        _format_opt(req, opt) for opt in destination_pool_options]
+        view_utils.format_opt(opt, keys) for opt in destination_pool_options]
     return {'destination_minion_pool_options': formatted_opts}
 
 
-def destination_options_collection(req, destination_options):
+def destination_options_collection(destination_options, keys=None):
     formatted_opts = [
-        _format_opt(req, opt) for opt in destination_options]
+        view_utils.format_opt(opt, keys) for opt in destination_options]
     return {'destination_options': formatted_opts}
 
 
-def source_minion_pool_options_collection(req, source_pool_options):
+def source_minion_pool_options_collection(source_pool_options, keys=None):
     formatted_opts = [
-        _format_opt(req, opt) for opt in source_pool_options]
+        view_utils.format_opt(opt, keys) for opt in source_pool_options]
     return {'source_minion_pool_options': formatted_opts}
 
 
-def source_options_collection(req, source_options):
+def source_options_collection(source_options, keys=None):
     formatted_opts = [
-        _format_opt(req, opt) for opt in source_options]
+        view_utils.format_opt(opt, keys) for opt in source_options]
     return {'source_options': formatted_opts}

coriolis/api/v1/views/endpoint_resources_view.py

@@ -1,38 +1,29 @@
 # Copyright 2020 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_resource(req, resource, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
+def instance_single(instance, keys=None):
+    return {"instance": view_utils.format_opt(instance, keys)}
 
-    return dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in resource.items()))
 
-
-def instance_single(req, instance):
-    return {"instance": _format_resource(req, instance)}
-
-
-def instances_collection(req, instances):
-    formatted_instances = [_format_resource(req, m)
+def instances_collection(instances, keys=None):
+    formatted_instances = [view_utils.format_opt(m, keys)
                            for m in instances]
     return {'instances': formatted_instances}
 
 
-def network_single(req, network):
-    return {"network": _format_resource(req, network)}
+def network_single(network, keys=None):
+    return {"network": view_utils.format_opt(network, keys)}
 
 
-def networks_collection(req, networks):
-    formatted_networks = [_format_resource(req, m) for m in networks]
+def networks_collection(networks, keys=None):
+    formatted_networks = [view_utils.format_opt(m, keys)
+                          for m in networks]
     return {'networks': formatted_networks}
 
 
-def storage_collection(req, storage):
-    formatted_storages = _format_resource(req, storage)
+def storage_collection(storage, keys=None):
+    formatted_storages = view_utils.format_opt(storage, keys)
     return {'storage': formatted_storages}

coriolis/api/v1/views/endpoint_view.py

@@ -1,17 +1,11 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_endpoint(req, endpoint, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
-    endpoint_dict = dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in endpoint.items()))
+def _format_endpoint(endpoint, keys=None):
+    endpoint_dict = view_utils.format_opt(endpoint, keys)
     mapped_regions = endpoint_dict.get('mapped_regions', [])
     endpoint_dict['mapped_regions'] = [
         reg['id'] for reg in mapped_regions]
@@ -19,11 +13,11 @@ def _format_endpoint(req, endpoint, keys=None):
     return endpoint_dict
 
 
-def single(req, endpoint):
-    return {"endpoint": _format_endpoint(req, endpoint)}
+def single(endpoint, keys=None):
+    return {"endpoint": _format_endpoint(endpoint, keys)}
 
 
-def collection(req, endpoints):
-    formatted_endpoints = [_format_endpoint(req, m)
+def collection(endpoints, keys=None):
+    formatted_endpoints = [_format_endpoint(m, keys)
                            for m in endpoints]
     return {'endpoints': formatted_endpoints}

coriolis/api/v1/views/migration_view.py

@@ -1,23 +1,16 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
-
 from coriolis.api.v1.views import replica_tasks_execution_view as view
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_migration(req, migration, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
-    migration_dict = dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in migration.items()))
+def _format_migration(migration, keys=None):
+    migration_dict = view_utils.format_opt(migration, keys)
 
     if len(migration_dict.get("executions", [])):
         execution = view.format_replica_tasks_execution(
-            req, migration_dict["executions"][0])
+            migration_dict["executions"][0], keys)
         del migration_dict["executions"]
     else:
         execution = {}
@@ -29,11 +22,11 @@ def _format_migration(req, migration, keys=None):
     return migration_dict
 
 
-def single(req, migration):
-    return {"migration": _format_migration(req, migration)}
+def single(migration, keys=None):
+    return {"migration": _format_migration(migration, keys)}
 
 
-def collection(req, migrations):
-    formatted_migrations = [_format_migration(req, m)
+def collection(migrations, keys=None):
+    formatted_migrations = [_format_migration(m, keys)
                             for m in migrations]
     return {'migrations': formatted_migrations}

coriolis/api/v1/views/minion_pool_view.py

@@ -1,17 +1,11 @@
 # Copyright 2020 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_minion_pool(req, minion_pool, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
-    minion_pool_dict = dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in minion_pool.items()))
+def _format_minion_pool(minion_pool, keys=None):
+    minion_pool_dict = view_utils.format_opt(minion_pool, keys)
 
     def _hide_minion_creds(minion_conn):
         if not minion_conn:
@@ -38,11 +32,11 @@ def _format_minion_pool(req, minion_pool, keys=None):
     return minion_pool_dict
 
 
-def single(req, minion_pool):
-    return {"minion_pool": _format_minion_pool(req, minion_pool)}
+def single(minion_pool, keys=None):
+    return {"minion_pool": _format_minion_pool(minion_pool, keys)}
 
 
-def collection(req, minion_pools):
+def collection(minion_pools, keys=None):
     formatted_minion_pools = [
-        _format_minion_pool(req, r) for r in minion_pools]
+        _format_minion_pool(r, keys) for r in minion_pools]
     return {'minion_pools': formatted_minion_pools}

coriolis/api/v1/views/region_view.py

@@ -1,17 +1,11 @@
 # Copyright 2020 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_region(req, region, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
-    region_dict = dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in region.items()))
+def _format_region(region, keys=None):
+    region_dict = view_utils.format_opt(region, keys)
 
     mapped_endpoints = region_dict.get('mapped_endpoints', [])
     region_dict['mapped_endpoints'] = [
@@ -24,11 +18,11 @@ def _format_region(req, region, keys=None):
     return region_dict
 
 
-def single(req, region):
-    return {"region": _format_region(req, region)}
+def single(region, keys=None):
+    return {"region": _format_region(region, keys)}
 
 
-def collection(req, regions):
+def collection(regions, keys=None):
     formatted_regions = [
-        _format_region(req, r) for r in regions]
+        _format_region(r, keys) for r in regions]
     return {'regions': formatted_regions}

coriolis/api/v1/views/replica_schedule_view.py

@@ -1,24 +1,14 @@
 # Copyright 2017 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
+from coriolis.api.v1.views import utils as view_utils
 
 
-def format_schedule(req, schedule, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
+def single(schedule, keys=None):
+    return {"schedule": view_utils.format_opt(schedule, keys)}
 
-    return dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in schedule.items()))
 
-
-def single(req, schedule):
-    return {"schedule": format_schedule(req, schedule)}
-
-
-def collection(req, schedules):
-    formatted_schedules = [format_schedule(req, m)
+def collection(schedules, keys=None):
+    formatted_schedules = [view_utils.format_opt(m, keys)
                            for m in schedules]
     return {'schedules': formatted_schedules}

coriolis/api/v1/views/replica_tasks_execution_view.py

@@ -1,10 +1,10 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
 
 from oslo_log import log as logging
 
+from coriolis.api.v1.views import utils as view_utils
 from coriolis import constants
 
 
@@ -12,7 +12,8 @@ LOG = logging.getLogger(__name__)
 
 
 def _sort_tasks(tasks, filter_error_only_tasks=True):
-    """ Sorts the given list of dicts representing tasks.
+    """
+    Sorts the given list of dicts representing tasks.
     Tasks are sorted primarily based on their index.
     """
     if filter_error_only_tasks:
@@ -24,26 +25,20 @@ def _sort_tasks(tasks, filter_error_only_tasks=True):
         tasks, key=lambda t: t.get('index', 0))
 
 
-def format_replica_tasks_execution(req, execution, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
+def format_replica_tasks_execution(execution, keys=None):
     if "tasks" in execution:
         execution["tasks"] = _sort_tasks(execution["tasks"])
 
-    execution_dict = dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in execution.items()))
+    execution_dict = view_utils.format_opt(execution, keys)
 
     return execution_dict
 
 
-def single(req, execution):
-    return {"execution": format_replica_tasks_execution(req, execution)}
+def single(execution, keys=None):
+    return {"execution": format_replica_tasks_execution(execution, keys)}
 
 
-def collection(req, executions):
-    formatted_executions = [format_replica_tasks_execution(req, m)
+def collection(executions, keys=None):
+    formatted_executions = [format_replica_tasks_execution(m, keys)
                             for m in executions]
     return {'executions': formatted_executions}

coriolis/api/v1/views/replica_view.py

@@ -1,33 +1,26 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
-
 from coriolis.api.v1.views import replica_tasks_execution_view as view
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_replica(req, replica, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
-    replica_dict = dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in replica.items()))
+def _format_replica(replica, keys=None):
+    replica_dict = view_utils.format_opt(replica, keys)
 
     executions = replica_dict.get('executions', [])
     replica_dict['executions'] = [
-        view.format_replica_tasks_execution(req, ex)
+        view.format_replica_tasks_execution(ex)
         for ex in executions]
 
     return replica_dict
 
 
-def single(req, replica):
-    return {"replica": _format_replica(req, replica)}
+def single(replica, keys=None):
+    return {"replica": _format_replica(replica, keys)}
 
 
-def collection(req, replicas):
-    formatted_replicas = [_format_replica(req, m)
+def collection(replicas, keys=None):
+    formatted_replicas = [_format_replica(m, keys)
                           for m in replicas]
     return {'replicas': formatted_replicas}

coriolis/api/v1/views/service_view.py

@@ -1,17 +1,11 @@
 # Copyright 2020 Cloudbase Solutions Srl
 # All Rights Reserved.
 
-import itertools
+from coriolis.api.v1.views import utils as view_utils
 
 
-def _format_service(req, service, keys=None):
-    def transform(key, value):
-        if keys and key not in keys:
-            return
-        yield (key, value)
-
-    service_dict = dict(itertools.chain.from_iterable(
-        transform(k, v) for k, v in service.items()))
+def _format_service(service, keys=None):
+    service_dict = view_utils.format_opt(service, keys)
 
     mapped_regions = service_dict.get('mapped_regions', [])
     service_dict['mapped_regions'] = [
@@ -20,11 +14,11 @@ def _format_service(req, service, keys=None):
     return service_dict
 
 
-def single(req, service):
-    return {"service": _format_service(req, service)}
+def single(service, keys=None):
+    return {"service": _format_service(service, keys)}
 
 
-def collection(req, services):
+def collection(services, keys=None):
     formatted_services = [
-        _format_service(req, r) for r in services]
+        _format_service(r, keys) for r in services]
     return {'services': formatted_services}

coriolis/api/v1/views/utils.py

@@ -0,0 +1,14 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import itertools
+
+
+def format_opt(option, keys=None):
+    def transform(key, value):
+        if keys and key not in keys:
+            return
+        yield (key, value)
+
+    return dict(itertools.chain.from_iterable(
+        transform(k, v) for k, v in option.items()))

coriolis/tests/api/v1/views/test_diagnostic_view.py

@@ -0,0 +1,53 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.api.v1.views import diagnostic_view
+from coriolis.tests import test_base
+
+
+class DiagnosticViewTestCase(test_base.CoriolisBaseTestCase):
+    "Test suite for the Coriolis api v1 views."""
+
+    def test_single(self):
+        mock_single = "mock_single"
+        expected_result = {'diagnostic': mock_single}
+
+        result = diagnostic_view.single(mock_single)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_single_none(self):
+        mock_single = None
+        expected_result = {'diagnostic': mock_single}
+
+        result = diagnostic_view.single(mock_single)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_collection(self):
+        mock_collection = {"mock_key_1": "value_1", "mock_key_2": "value_2"}
+        expected_result = {'diagnostics': mock_collection}
+
+        result = diagnostic_view.collection(mock_collection)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_collection_none(self):
+        mock_collection = {}
+        expected_result = {'diagnostics': mock_collection}
+
+        result = diagnostic_view.collection(mock_collection)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )

coriolis/tests/api/v1/views/test_endpoint_options_view.py

@@ -0,0 +1,28 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.api.v1.views import endpoint_options_view
+from coriolis.tests import test_base
+
+
+class EndpointOptionsViewTestCase(test_base.CoriolisApiViewsTestCase):
+
+    def test_destination_minion_pool_options_collection(self):
+        fun = getattr(endpoint_options_view,
+                      'destination_minion_pool_options_collection')
+        self._collection_view_test(fun, "destination_minion_pool_options")
+
+    def test_destination_options_collection(self):
+        fun = getattr(endpoint_options_view,
+                      'destination_options_collection')
+        self._collection_view_test(fun, "destination_options")
+
+    def test_source_minion_pool_options_collection(self):
+        fun = getattr(endpoint_options_view,
+                      'source_minion_pool_options_collection')
+        self._collection_view_test(fun, "source_minion_pool_options")
+
+    def test_source_options_collection(self):
+        fun = getattr(endpoint_options_view,
+                      'source_options_collection')
+        self._collection_view_test(fun, "source_options")

coriolis/tests/api/v1/views/test_endpoint_resources_view.py

@@ -0,0 +1,33 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.api.v1.views import endpoint_resources_view
+from coriolis.tests import test_base
+
+
+class EndpointResourcesViewTestCase(test_base.CoriolisApiViewsTestCase):
+
+    def test_instance_single(self):
+        fun = getattr(endpoint_resources_view,
+                      'instance_single')
+        self._single_view_test(fun, "instance")
+
+    def test_instances_collection(self):
+        fun = getattr(endpoint_resources_view,
+                      'instances_collection')
+        self._collection_view_test(fun, "instances")
+
+    def test_network_single(self):
+        fun = getattr(endpoint_resources_view,
+                      'network_single')
+        self._single_view_test(fun, "network")
+
+    def test_networks_collection(self):
+        fun = getattr(endpoint_resources_view,
+                      'networks_collection')
+        self._collection_view_test(fun, "networks")
+
+    def test_storage_collection(self):
+        fun = getattr(endpoint_resources_view,
+                      'storage_collection')
+        self._single_view_test(fun, "storage")

coriolis/tests/api/v1/views/test_endpoint_view.py

@@ -0,0 +1,87 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1.views import endpoint_view
+from coriolis.api.v1.views import utils as view_utils
+from coriolis.tests import test_base
+
+
+class EndpointViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    def setUp(self):
+        super(EndpointViewTestCase, self).setUp()
+        self._format_fun = endpoint_view._format_endpoint
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_endpoint(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mapped_regions": [{'id': 'mock_id1'}, {'id': 'mock_id2'}],
+                "mock_key": "mock_value"
+            }
+
+            expected_result = {
+                'mapped_regions': ['mock_id1', 'mock_id2'],
+                'mock_key': 'mock_value'
+            }
+
+            endpoint = mock.sentinel.endpoint
+            keys = mock.sentinel.keys
+            result = endpoint_view._format_endpoint(endpoint, keys)
+
+            mock_format_opt.assert_called_once_with(endpoint, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_endpoint_no_keys(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mapped_regions": [{'id': 'mock_id1'}, {'id': 'mock_id2'}],
+            }
+
+            expected_result = {
+                'mapped_regions': ['mock_id1', 'mock_id2'],
+            }
+
+            endpoint = mock.sentinel.endpoint
+            keys = mock.sentinel.keys
+            result = endpoint_view._format_endpoint(endpoint, keys)
+
+            mock_format_opt.assert_called_once_with(endpoint, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_endpoint_no_mapped_regions(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mock_key": "mock_value"
+            }
+
+            expected_result = {
+                'mapped_regions': [],
+                'mock_key': 'mock_value'
+            }
+
+            endpoint = mock.sentinel.endpoint
+            keys = mock.sentinel.keys
+            result = endpoint_view._format_endpoint(endpoint, keys)
+
+            mock_format_opt.assert_called_once_with(endpoint, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    def test_single(self):
+        fun = getattr(endpoint_view, 'single')
+        self._single_view_test(fun, 'endpoint')
+
+    def test_collection(self):
+        fun = getattr(endpoint_view, 'collection')
+        self._collection_view_test(fun, 'endpoints')

coriolis/tests/api/v1/views/test_migration_view.py

@@ -0,0 +1,96 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1.views import migration_view
+from coriolis.api.v1.views import replica_tasks_execution_view as view
+from coriolis.api.v1.views import utils as view_utils
+from coriolis.tests import test_base
+
+
+class MigrationViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    @mock.patch.object(view, 'format_replica_tasks_execution')
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_migration(
+        self,
+        mock_format_opt,
+        mock_format_replica_tasks_execution
+    ):
+        mock_execution = {'tasks': 'mock_id1'}
+        mock_format_opt.return_value = {
+            "executions": [mock_execution],
+            'tasks': 'mock_id2',
+            'mock_key': 'mock_value'
+        }
+        mock_format_replica_tasks_execution.return_value = mock_execution
+
+        expected_result = {
+            'tasks': 'mock_id1',
+            'mock_key': 'mock_value'
+        }
+
+        endpoint = mock.sentinel.endpoint
+        keys = mock.sentinel.keys
+        result = migration_view._format_migration(endpoint, keys)
+
+        mock_format_replica_tasks_execution.assert_called_once_with(
+            mock_execution, keys
+        )
+        mock_format_opt.assert_called_once_with(endpoint, keys)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_migration_no_tasks(
+        self,
+        mock_format_opt,
+    ):
+        mock_format_opt.return_value = {
+            'mock_key': 'mock_value'
+        }
+
+        endpoint = mock.sentinel.endpoint
+        keys = mock.sentinel.keys
+        result = migration_view._format_migration(endpoint, keys)
+
+        mock_format_opt.assert_called_once_with(endpoint, keys)
+
+        self.assertEqual(
+            mock_format_opt.return_value,
+            result
+        )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_migration_migration_dict_has_tasks(
+        self,
+        mock_format_opt,
+    ):
+        mock_format_opt.return_value = {
+            'tasks': 'mock_id1',
+            'mock_key': 'mock_value'
+        }
+
+        endpoint = mock.sentinel.endpoint
+        keys = mock.sentinel.keys
+        result = migration_view._format_migration(endpoint, keys)
+
+        mock_format_opt.assert_called_once_with(endpoint, keys)
+
+        self.assertEqual(
+            mock_format_opt.return_value,
+            result
+        )
+
+    def test_single(self):
+        fun = getattr(migration_view, 'single')
+        self._single_view_test(fun, 'migration')
+
+    def test_collection(self):
+        fun = getattr(migration_view, 'collection')
+        self._collection_view_test(fun, 'migrations')

coriolis/tests/api/v1/views/test_minion_pool_view.py

@@ -0,0 +1,120 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1.views import minion_pool_view as view
+from coriolis.api.v1.views import utils as view_utils
+from coriolis.tests import test_base
+
+
+class MinionPoolViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_minion_pool(
+        self,
+        mock_format_opt,
+    ):
+        mock_minion_pool_dict = {
+            'minion_machines': [{
+                'connection_info': {
+                    'pkey': 'mock_key',
+                    'password': 'mock_key',
+                    'certificates': {'key': 'mock_key'}
+                },
+                'backup_writer_connection_info': {
+                    'connection_details': {
+                        'pkey': 'mock_key',
+                        'password': 'mock_key',
+                        'certificates': {'key': 'mock_key'}
+                    },
+                }
+            }],
+        }
+        expected_result = {
+            'minion_machines': [{
+                'connection_info': {
+                    'pkey': '***',
+                    'password': '***',
+                    'certificates': {'key': '***'}
+                },
+                'backup_writer_connection_info': {
+                    'connection_details': {
+                        'pkey': '***',
+                        'password': '***',
+                        'certificates': {'key': '***'}
+                    },
+                }
+            }],
+        }
+        mock_format_opt.return_value = mock_minion_pool_dict
+
+        endpoint = mock.sentinel.endpoint
+        keys = mock.sentinel.keys
+        result = view._format_minion_pool(endpoint, keys)
+
+        mock_format_opt.assert_called_once_with(endpoint, keys)
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_minion_pool_connection_none(
+        self,
+        mock_format_opt,
+    ):
+        mock_minion_pool_dict = {
+            'minion_machines': [{
+                'connection_info': None,
+                'backup_writer_connection_info': {
+                    'connection_details': None,
+                }
+            }],
+        }
+        expected_result = {
+            'minion_machines': [{
+                'connection_info': None,
+                'backup_writer_connection_info': {
+                    'connection_details': None,
+                }
+            }],
+        }
+        mock_format_opt.return_value = mock_minion_pool_dict
+
+        endpoint = mock.sentinel.endpoint
+        keys = mock.sentinel.keys
+        result = view._format_minion_pool(endpoint, keys)
+
+        mock_format_opt.assert_called_once_with(endpoint, keys)
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_minion_pool_no_minion_machines(
+        self,
+        mock_format_opt,
+    ):
+        expected_result = {}
+        mock_format_opt.return_value = {}
+
+        endpoint = mock.sentinel.endpoint
+        keys = mock.sentinel.keys
+        result = view._format_minion_pool(endpoint, keys)
+
+        mock_format_opt.assert_called_once_with(endpoint, keys)
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_single(self):
+        fun = getattr(view, 'single')
+        self._single_view_test(fun, 'minion_pool')
+
+    def test_collection(self):
+        fun = getattr(view, 'collection')
+        self._collection_view_test(fun, 'minion_pools')

coriolis/tests/api/v1/views/test_region_view.py

@@ -0,0 +1,96 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1.views import region_view
+from coriolis.api.v1.views import utils as view_utils
+from coriolis.tests import test_base
+
+
+class RegionViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    def setUp(self):
+        super(RegionViewTestCase, self).setUp()
+        self._format_fun = region_view._format_region
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_region(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mapped_endpoints": [{'id': 'endpoint_1'},
+                                     {'id': 'endpoint_2'}],
+                "mapped_services": [{'id': 'service_1'},
+                                    {'id': 'service_2'}],
+                "mock_key": "mock_value"
+            }
+
+            expected_result = {
+                'mapped_endpoints': ['endpoint_1', 'endpoint_2'],
+                "mapped_services": ['service_1', 'service_2'],
+                'mock_key': 'mock_value'
+            }
+
+            region = mock.sentinel.region
+            keys = mock.sentinel.keys
+            result = region_view._format_region(region, keys)
+
+            mock_format_opt.assert_called_once_with(region, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_region_no_keys(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                'mapped_endpoints': [{'id': 'endpoint_1'},
+                                     {'id': 'endpoint_2'}],
+                'mapped_services': [{'id': 'service_1'},
+                                    {'id': 'service_2'}],
+            }
+
+            expected_result = {
+                'mapped_endpoints': ['endpoint_1', 'endpoint_2'],
+                'mapped_services': ['service_1', 'service_2'],
+            }
+
+            region = mock.sentinel.region
+            keys = mock.sentinel.keys
+            result = region_view._format_region(region, keys)
+
+            mock_format_opt.assert_called_once_with(region, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_region_no_mapped_regions(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mock_key": "mock_value"
+            }
+
+            expected_result = {
+                'mapped_endpoints': [],
+                'mapped_services': [],
+                'mock_key': 'mock_value'
+            }
+
+            region = mock.sentinel.region
+            keys = mock.sentinel.keys
+            result = region_view._format_region(region, keys)
+
+            mock_format_opt.assert_called_once_with(region, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    def test_single(self):
+        fun = getattr(region_view, 'single')
+        self._single_view_test(fun, 'region')
+
+    def test_collection(self):
+        fun = getattr(region_view, 'collection')
+        self._collection_view_test(fun, 'regions')

coriolis/tests/api/v1/views/test_replica_schedule_view.py

@@ -0,0 +1,17 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.api.v1.views import replica_schedule_view
+from coriolis.tests import test_base
+
+
+class ReplicaViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    def test_single(self):
+        fun = getattr(replica_schedule_view, 'single')
+        self._single_view_test(fun, 'schedule')
+
+    def test_collection(self):
+        fun = getattr(replica_schedule_view, 'collection')
+        self._collection_view_test(fun, 'schedules')

coriolis/tests/api/v1/views/test_replica_task_execution_view.py

@@ -0,0 +1,117 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1.views import replica_tasks_execution_view as view
+from coriolis.api.v1.views import utils as view_utils
+from coriolis import constants
+from coriolis.tests import test_base
+
+
+class ReplicaTaskExecutionViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    @mock.patch.object(view, '_sort_tasks')
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_replica_tasks_execution(
+        self,
+        mock_format_opt,
+        mock_sort_tasks
+    ):
+        mock_tasks = ['mock_task1', 'mock_task2']
+        mock_execution = {
+            'tasks': mock_tasks,
+            'mock_key': 'mock_value'
+        }
+        mock_sort_tasks.return_value = mock_execution
+
+        keys = mock.sentinel.keys
+        result = view.format_replica_tasks_execution(mock_execution, keys)
+
+        mock_sort_tasks.assert_called_once_with(mock_tasks)
+        mock_format_opt.assert_called_once_with(mock_execution["tasks"], keys)
+        self.assertEqual(
+            mock_format_opt.return_value,
+            result
+        )
+
+    @mock.patch.object(view, '_sort_tasks')
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_replica_tasks_execution_no_tasks(
+        self,
+        mock_format_opt,
+        mock_sort_tasks
+    ):
+        mock_execution = {
+            'mock_key': 'mock_value'
+        }
+
+        keys = mock.sentinel.keys
+        result = view.format_replica_tasks_execution(mock_execution, keys)
+
+        mock_sort_tasks.assert_not_called()
+        mock_format_opt.assert_called_once_with(mock_execution, keys)
+        self.assertEqual(
+            mock_format_opt.return_value,
+            result
+        )
+
+    def test_sort_tasks(self):
+        mock_tasks = [
+            {'index': 2, 'status': 'mock_status1'},
+            {'status': 'mock_status3'},
+            {'index': 3, 'status': constants.TASK_STATUS_ON_ERROR_ONLY},
+            {'index': 1, 'status': 'mock_status2'},
+        ]
+        expected_result = [
+            {'status': 'mock_status3'},
+            {'index': 1, 'status': 'mock_status2'},
+            {'index': 2, 'status': 'mock_status1'},
+        ]
+
+        result = view._sort_tasks(mock_tasks)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_sort_tasks_no_filter(self):
+        mock_tasks = [
+            {'index': 2, 'status': 'mock_status1'},
+            {'status': 'mock_status3'},
+            {'index': 3, 'status': constants.TASK_STATUS_ON_ERROR_ONLY},
+            {'index': 1, 'status': 'mock_status2'},
+        ]
+        expected_result = [
+            {'status': 'mock_status3'},
+            {'index': 1, 'status': 'mock_status2'},
+            {'index': 2, 'status': 'mock_status1'},
+            {'index': 3, 'status': constants.TASK_STATUS_ON_ERROR_ONLY},
+        ]
+
+        result = view._sort_tasks(mock_tasks, False)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_sort_tasks_no_tasks(self):
+        expected_result = []
+
+        result = view._sort_tasks(expected_result)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_single(self):
+        fun = getattr(view, 'single')
+        self._single_view_test(fun, 'execution')
+
+    def test_collection(self):
+        fun = getattr(view, 'collection')
+        self._collection_view_test(fun, 'executions')

coriolis/tests/api/v1/views/test_replica_view.py

@@ -0,0 +1,112 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1.views import replica_tasks_execution_view as view
+from coriolis.api.v1.views import replica_view
+from coriolis.api.v1.views import utils as view_utils
+from coriolis.tests import test_base
+
+
+class ReplicaViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    def setUp(self):
+        super(ReplicaViewTestCase, self).setUp()
+        self._format_fun = replica_view._format_replica
+
+    @mock.patch.object(view, 'format_replica_tasks_execution')
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_replica(self, mock_format_opt,
+                            mock_format_replica_tasks_execution):
+            mock_format_opt.return_value = {
+                "executions": [{'id': 'mock_id1'}, {'id': 'mock_id2'}],
+                "mock_key": "mock_value"
+            }
+
+            expected_calls = [
+                mock.call.mock_format_replica_tasks_execution(
+                    {'id': 'mock_id1'}),
+                mock.call.mock_format_replica_tasks_execution(
+                    {'id': 'mock_id2'})]
+            expected_result = {
+                "executions":
+                    [mock_format_replica_tasks_execution.return_value,
+                     mock_format_replica_tasks_execution.return_value],
+                'mock_key': 'mock_value'
+            }
+
+            replica = mock.sentinel.replica
+            keys = mock.sentinel.keys
+            result = replica_view._format_replica(replica, keys)
+
+            mock_format_opt.assert_called_once_with(replica, keys)
+            mock_format_replica_tasks_execution.assert_has_calls(
+                expected_calls
+            )
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view, 'format_replica_tasks_execution')
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_replica_no_keys(self, mock_format_opt,
+                                    mock_format_replica_tasks_execution):
+            mock_format_opt.return_value = {
+                "executions": [{'id': 'mock_id1'}, {'id': 'mock_id2'}],
+            }
+
+            expected_calls = [
+                mock.call.mock_format_replica_tasks_execution(
+                    {'id': 'mock_id1'}),
+                mock.call.mock_format_replica_tasks_execution(
+                    {'id': 'mock_id2'})]
+            expected_result = {
+                "executions":
+                    [mock_format_replica_tasks_execution.return_value,
+                     mock_format_replica_tasks_execution.return_value],
+            }
+
+            replica = mock.sentinel.replica
+            keys = mock.sentinel.keys
+            result = replica_view._format_replica(replica, keys)
+
+            mock_format_opt.assert_called_once_with(replica, keys)
+            mock_format_replica_tasks_execution.assert_has_calls(
+                expected_calls
+            )
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_replica_no_executions(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mock_key": "mock_value"
+            }
+
+            expected_result = {
+                'executions': [],
+                'mock_key': 'mock_value'
+            }
+
+            replica = mock.sentinel.replica
+            keys = mock.sentinel.keys
+            result = replica_view._format_replica(replica, keys)
+
+            mock_format_opt.assert_called_once_with(replica, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    def test_single(self):
+        fun = getattr(replica_view, 'single')
+        self._single_view_test(fun, 'replica')
+
+    def test_collection(self):
+        fun = getattr(replica_view, 'collection')
+        self._collection_view_test(fun, 'replicas')

coriolis/tests/api/v1/views/test_service_view.py

@@ -0,0 +1,87 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1.views import service_view
+from coriolis.api.v1.views import utils as view_utils
+from coriolis.tests import test_base
+
+
+class ServiceViewTestCase(test_base.CoriolisApiViewsTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    def setUp(self):
+        super(ServiceViewTestCase, self).setUp()
+        self._format_fun = service_view._format_service
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_service(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mapped_regions": [{'id': 'mock_id1'}, {'id': 'mock_id2'}],
+                "mock_key": "mock_value"
+            }
+
+            expected_result = {
+                "mapped_regions": ['mock_id1', 'mock_id2'],
+                'mock_key': 'mock_value'
+            }
+
+            service = mock.sentinel.service
+            keys = mock.sentinel.keys
+            result = service_view._format_service(service, keys)
+
+            mock_format_opt.assert_called_once_with(service, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_service_no_keys(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mapped_regions": [{'id': 'mock_id1'}, {'id': 'mock_id2'}],
+            }
+
+            expected_result = {
+                "mapped_regions": ['mock_id1', 'mock_id2'],
+            }
+
+            service = mock.sentinel.service
+            keys = mock.sentinel.keys
+            result = service_view._format_service(service, keys)
+
+            mock_format_opt.assert_called_once_with(service, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    @mock.patch.object(view_utils, 'format_opt')
+    def test_format_service_no_mapped_regions(self, mock_format_opt):
+            mock_format_opt.return_value = {
+                "mock_key": "mock_value"
+            }
+
+            expected_result = {
+                'mapped_regions': [],
+                'mock_key': 'mock_value'
+            }
+
+            service = mock.sentinel.service
+            keys = mock.sentinel.keys
+            result = service_view._format_service(service, keys)
+
+            mock_format_opt.assert_called_once_with(service, keys)
+            self.assertEqual(
+                expected_result,
+                result
+            )
+
+    def test_single(self):
+        fun = getattr(service_view, 'single')
+        self._single_view_test(fun, 'service')
+
+    def test_collection(self):
+        fun = getattr(service_view, 'collection')
+        self._collection_view_test(fun, 'services')

coriolis/tests/api/v1/views/test_utils.py

@@ -0,0 +1,48 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.api.v1.views import utils as view_utils
+from coriolis.tests import test_base
+
+
+class ViewUtilsTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis api v1 views."""
+
+    def test_format_opt(self):
+        mock_option = {"mock_key_1": "value_1", "mock_key_2": "value_2"}
+        mock_keys = {"mock_key_1"}
+
+        expected_result = {"mock_key_1": "value_1"}
+
+        result = view_utils.format_opt(mock_option, mock_keys)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_format_opt_key_not_in_options(self):
+        mock_option = {"mock_key_1": "value_1", "mock_key_2": "value_2"}
+        mock_keys = {"mock_key_3"}
+
+        expected_result = {}
+
+        result = view_utils.format_opt(mock_option, mock_keys)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )
+
+    def test_format_opt_keys_none(self):
+        mock_option = {"mock_key_1": "value_1", "mock_key_2": "value_2"}
+        mock_keys = None
+
+        expected_result = mock_option
+
+        result = view_utils.format_opt(mock_option, mock_keys)
+
+        self.assertEqual(
+            expected_result,
+            result
+        )

coriolis/tests/test_base.py

@@ -3,10 +3,46 @@
 
 """Defines base class for all tests."""
 
+from unittest import mock
+
 from oslotest import base
 
+from coriolis.api.v1.views import utils as views_utils
+
 
 class CoriolisBaseTestCase(base.BaseTestCase):
 
     def setUp(self):
         super(CoriolisBaseTestCase, self).setUp()
+
+
+class CoriolisApiViewsTestCase(CoriolisBaseTestCase):
+
+    def setUp(self):
+        super(CoriolisApiViewsTestCase, self).setUp()
+        self._single_response = {"key1": "value1", "key2": "value2"}
+        self._collection_response = [
+            self._single_response, self._single_response]
+        self._format_fun = views_utils.format_opt
+
+    def _single_view_test(self, fun, expected_result_key, keys=None):
+        format_fun = '%s.%s' % (
+            self._format_fun.__module__, self._format_fun.__name__)
+        with mock.patch(format_fun) as format_mock:
+            result = fun(self._single_response, keys)
+            format_mock.assert_called_once_with(self._single_response, keys)
+            expected_result = {expected_result_key: format_mock.return_value}
+            self.assertEqual(expected_result, result)
+
+    def _collection_view_test(
+            self, fun, expected_result_key, keys=None):
+        format_fun = '%s.%s' % (
+            self._format_fun.__module__, self._format_fun.__name__)
+        with mock.patch(format_fun) as format_mock:
+            f_opts = []
+            result = fun(self._collection_response, keys)
+            for c in self._collection_response:
+                format_mock.assert_has_calls([mock.call(c, keys)])
+                f_opts.append(format_mock.return_value)
+            expected_result = {expected_result_key: f_opts}
+            self.assertEqual(expected_result, result)