1  """Tests for MiddlewareRegistry resource sharing service binding."""
  2  import pytest
  3  from solace_agent_mesh.common.middleware.registry import MiddlewareRegistry
  4  from solace_agent_mesh.common.services.default_resource_sharing_service import (
  5      DefaultResourceSharingService,
  6  )
  7  from solace_agent_mesh.services.resource_sharing_service import (
  8      ResourceSharingService,
  9      ResourceType,
 10  )
 11  
 12  
 13  class TestResourceSharingServiceBinding:
 14      """Test resource sharing service binding in MiddlewareRegistry."""
 15  
 16      def setup_method(self):
 17          """Reset registry before each test."""
 18          MiddlewareRegistry.reset_bindings()
 19  
 20      def teardown_method(self):
 21          """Reset registry after each test."""
 22          MiddlewareRegistry.reset_bindings()
 23  
 24      def test_get_resource_sharing_service_returns_default_when_none_bound(self):
 25          """Test that get_resource_sharing_service returns default when none bound."""
 26          service_class = MiddlewareRegistry.get_resource_sharing_service()
 27          assert service_class == DefaultResourceSharingService
 28  
 29      def test_bind_resource_sharing_service_stores_custom_service(self):
 30          """Test that bind_resource_sharing_service stores custom service."""
 31  
 32          class CustomResourceSharingService(ResourceSharingService):
 33              def get_shared_resource_ids(self, session, user_email, resource_type):
 34                  return ["custom-id"]
 35  
 36              def check_user_access(
 37                  self, session, resource_id, resource_type, user_email
 38              ):
 39                  return "CUSTOM_ACCESS"
 40  
 41              def delete_resource_shares(self, session, resource_id, resource_type):
 42                  return True
 43  
 44          MiddlewareRegistry.bind_resource_sharing_service(CustomResourceSharingService)
 45          service_class = MiddlewareRegistry.get_resource_sharing_service()
 46          assert service_class == CustomResourceSharingService
 47  
 48      def test_bind_resource_sharing_service_replaces_existing_binding(self):
 49          """Test that binding a new service replaces the existing one."""
 50  
 51          class FirstService(ResourceSharingService):
 52              def get_shared_resource_ids(self, session, user_email, resource_type):
 53                  return []
 54  
 55              def check_user_access(
 56                  self, session, resource_id, resource_type, user_email
 57              ):
 58                  return None
 59  
 60              def delete_resource_shares(self, session, resource_id, resource_type):
 61                  return True
 62  
 63          class SecondService(ResourceSharingService):
 64              def get_shared_resource_ids(self, session, user_email, resource_type):
 65                  return []
 66  
 67              def check_user_access(
 68                  self, session, resource_id, resource_type, user_email
 69              ):
 70                  return None
 71  
 72              def delete_resource_shares(self, session, resource_id, resource_type):
 73                  return True
 74  
 75          MiddlewareRegistry.bind_resource_sharing_service(FirstService)
 76          MiddlewareRegistry.bind_resource_sharing_service(SecondService)
 77          service_class = MiddlewareRegistry.get_resource_sharing_service()
 78          assert service_class == SecondService
 79  
 80      def test_reset_bindings_clears_resource_sharing_service(self):
 81          """Test that reset_bindings clears resource sharing service binding."""
 82  
 83          class CustomService(ResourceSharingService):
 84              def get_shared_resource_ids(self, session, user_email, resource_type):
 85                  return []
 86  
 87              def check_user_access(
 88                  self, session, resource_id, resource_type, user_email
 89              ):
 90                  return None
 91  
 92              def delete_resource_shares(self, session, resource_id, resource_type):
 93                  return True
 94  
 95          MiddlewareRegistry.bind_resource_sharing_service(CustomService)
 96          assert MiddlewareRegistry.get_resource_sharing_service() == CustomService
 97  
 98          MiddlewareRegistry.reset_bindings()
 99          service_class = MiddlewareRegistry.get_resource_sharing_service()
100          assert service_class == DefaultResourceSharingService
101  
102      def test_get_registry_status_includes_resource_sharing_service(self):
103          """Test that get_registry_status includes resource sharing service info."""
104          status = MiddlewareRegistry.get_registry_status()
105          assert "resource_sharing_service" in status
106          assert status["resource_sharing_service"] == "default"
107  
108      def test_get_registry_status_shows_custom_resource_sharing_service(self):
109          """Test that get_registry_status shows custom service name."""
110  
111          class CustomService(ResourceSharingService):
112              def get_shared_resource_ids(self, session, user_email, resource_type):
113                  return []
114  
115              def check_user_access(
116                  self, session, resource_id, resource_type, user_email
117              ):
118                  return None
119  
120              def delete_resource_shares(self, session, resource_id, resource_type):
121                  return True
122  
123          MiddlewareRegistry.bind_resource_sharing_service(CustomService)
124          status = MiddlewareRegistry.get_registry_status()
125          assert status["resource_sharing_service"] == "CustomService"
126  
127      def test_get_registry_status_has_custom_bindings_with_resource_sharing(self):
128          """Test that has_custom_bindings is True when resource sharing service is bound."""
129  
130          class CustomService(ResourceSharingService):
131              def get_shared_resource_ids(self, session, user_email, resource_type):
132                  return []
133  
134              def check_user_access(
135                  self, session, resource_id, resource_type, user_email
136              ):
137                  return None
138  
139              def delete_resource_shares(self, session, resource_id, resource_type):
140                  return True
141  
142          MiddlewareRegistry.bind_resource_sharing_service(CustomService)
143          status = MiddlewareRegistry.get_registry_status()
144          assert status["has_custom_bindings"] is True