Merge pull request #43 from alexander-zuev/fix/api-spec

fix(api): fix a bug with api_spec retrieval

Author: alexander-zuev

.github/workflows/docs/release-checklist.md

@@ -30,7 +30,7 @@ Pre-release
 9. Update .env.example (if necessary) - [X]
 
 Post-release
-10. Clean install from PyPi works
+10. Clean install from PyPi works - [X]
 
 
 

supabase_mcp/services/api/spec_manager.py

@@ -9,7 +9,7 @@
 
 # Constants
 SPEC_URL = "https://api.supabase.com/api/v1-json"
-LOCAL_SPEC_PATH = Path(__file__).parent / "services" / "api" / "specs" / "api_spec.json"
+LOCAL_SPEC_PATH = Path(__file__).parent / "specs" / "api_spec.json"
 
 
 class ApiDomain(str, Enum):
@@ -212,44 +212,50 @@ def _build_caches(self) -> None:
 
 
 # Example usage (assuming you have an instance of ApiSpecManager called 'spec_manager'):
-async def main():
-    spec_manager = await ApiSpecManager.create()
+async def main() -> None:
+    """Test function to demonstrate ApiSpecManager functionality."""
+    # Create a new instance of ApiSpecManager
     spec_manager = ApiSpecManager()
-    spec_manager.spec = json.loads(LOCAL_SPEC_PATH)  # Load your JSON here for testing
-    spec_manager._build_caches()
 
-    # 1. Get all paths and methods
+    # Load the spec
+    await spec_manager.get_spec()
+
+    # Print the path to help debug
+    print(f"Looking for spec at: {LOCAL_SPEC_PATH}")
+
+    # 1. Get all domains
+    all_domains = spec_manager.get_all_domains()
+    print("\nAll Domains:")
+    print(all_domains)
+
+    # 2. Get all paths and methods
     all_paths = spec_manager.get_all_paths_and_methods()
-    print("All Paths and Methods:")
-    for path, methods in all_paths.items():
+    print("\nAll Paths and Methods (sample):")
+    # Just print a few to avoid overwhelming output
+    for i, (path, methods) in enumerate(all_paths.items()):
+        if i >= 5:  # Limit to 5 paths
+            break
         print(f"  {path}:")
         for method, operation_id in methods.items():
             print(f"    {method}: {operation_id}")
 
-    # 2. Get paths and methods for the "Auth" domain
-    auth_paths = spec_manager.get_paths_and_methods_by_domain("Auth")
-    print("\nAuth Paths and Methods:")
-    for path, methods in auth_paths.items():
+    # 3. Get paths and methods for the "Edge Functions" domain
+    edge_paths = spec_manager.get_paths_and_methods_by_domain("Edge Functions")
+    print("\nEdge Functions Paths and Methods:")
+    for path, methods in edge_paths.items():
         print(f"  {path}:")
         for method, operation_id in methods.items():
             print(f"    {method}: {operation_id}")
 
-    # 3. Get all domains
-    all_domains = spec_manager.get_all_domains()
-    print("\nAll Domains:")
-    print(all_domains)
-
     # 4. Get the full spec for a specific path and method
-    path = "/v1/projects/{ref}/config/auth"
+    path = "/v1/projects/{ref}/functions"
     method = "GET"
     full_spec = spec_manager.get_spec_for_path_and_method(path, method)
     print(f"\nFull Spec for {method} {path}:")
-    print(json.dumps(full_spec, indent=2))
-
-    # # Example of get_spec_part
-    # schemas = spec_manager.get_spec_part("components", "schemas", "AuthConfigResponse")
-    # print("\nAuthConfigResponse schema:")
-    # print(json.dumps(schemas, indent=2))
+    if full_spec:
+        print(json.dumps(full_spec, indent=2)[:500] + "...")  # Truncate for readability
+    else:
+        print("Spec not found for this path/method")
 
 
 if __name__ == "__main__":

tests/services/api/test_spec_manager.py

@@ -165,3 +165,114 @@ async def test_get_spec_not_loaded(self, spec_manager_integration: ApiSpecManage
 
         assert result == SAMPLE_SPEC
         mock_fetch.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_comprehensive_spec_retrieval(self, spec_manager_integration: ApiSpecManager):
+        """
+        Comprehensive test of API spec retrieval and functionality.
+        This test exactly mirrors the main() function to ensure all aspects work correctly.
+        """
+        # Create a fresh instance to avoid any cached data from other tests
+        from supabase_mcp.services.api.spec_manager import LOCAL_SPEC_PATH, ApiSpecManager
+
+        spec_manager = ApiSpecManager()
+
+        # Print the path being used (for debugging)
+        print(f"\nTest is looking for spec at: {LOCAL_SPEC_PATH}")
+
+        # Load the spec
+        spec = await spec_manager.get_spec()
+        assert spec is not None, "Spec should be loaded successfully"
+
+        # 1. Test get_all_domains
+        all_domains = spec_manager.get_all_domains()
+        print(f"\nAll domains: {all_domains}")
+        assert len(all_domains) > 0, "Should have at least one domain"
+
+        # Verify all expected domains are present
+        expected_domains = [
+            "Analytics",
+            "Auth",
+            "Database",
+            "Domains",
+            "Edge Functions",
+            "Environments",
+            "OAuth",
+            "Organizations",
+            "Projects",
+            "Rest",
+            "Secrets",
+            "Storage",
+        ]
+        for domain in expected_domains:
+            assert domain in all_domains, f"Domain '{domain}' should be in the list of domains"
+
+        # 2. Test get_all_paths_and_methods
+        all_paths = spec_manager.get_all_paths_and_methods()
+        assert len(all_paths) > 0, "Should have at least one path"
+
+        # Sample a few paths to verify
+        sample_paths = list(all_paths.keys())[:5]
+        print("\nSample paths:")
+        for path in sample_paths:
+            print(f"  {path}:")
+            assert path.startswith("/v1/"), f"Path {path} should start with /v1/"
+            assert len(all_paths[path]) > 0, f"Path {path} should have at least one method"
+            for method, operation_id in all_paths[path].items():
+                print(f"    {method}: {operation_id}")
+                assert method.lower() in ["get", "post", "put", "patch", "delete"], f"Method {method} should be valid"
+                assert operation_id.startswith("v1-"), f"Operation ID {operation_id} should start with v1-"
+
+        # 3. Test get_paths_and_methods_by_domain for each domain
+        for domain in expected_domains:
+            domain_paths = spec_manager.get_paths_and_methods_by_domain(domain)
+            assert len(domain_paths) > 0, f"Domain {domain} should have at least one path"
+            print(f"\n{domain} domain has {len(domain_paths)} paths")
+
+        # 4. Test Edge Functions domain specifically
+        edge_paths = spec_manager.get_paths_and_methods_by_domain("Edge Functions")
+        print("\nEdge Functions Paths and Methods:")
+        for path in edge_paths:
+            print(f"  {path}")
+            for method, operation_id in edge_paths[path].items():
+                print(f"    {method}: {operation_id}")
+
+        # Verify specific Edge Functions paths exist
+        expected_edge_paths = [
+            "/v1/projects/{ref}/functions",
+            "/v1/projects/{ref}/functions/{function_slug}",
+            "/v1/projects/{ref}/functions/deploy",
+        ]
+        for path in expected_edge_paths:
+            assert path in edge_paths, f"Expected path {path} should be in Edge Functions domain"
+
+        # 5. Test get_spec_for_path_and_method
+        # Test for Edge Functions
+        path = "/v1/projects/{ref}/functions"
+        method = "GET"
+        full_spec = spec_manager.get_spec_for_path_and_method(path, method)
+        assert full_spec is not None, f"Should find spec for {method} {path}"
+        assert "operationId" in full_spec, "Spec should include operationId"
+        assert full_spec["operationId"] == "v1-list-all-functions", "Should have correct operationId"
+
+        # Test for another domain (Auth)
+        auth_path = "/v1/projects/{ref}/config/auth"
+        auth_method = "GET"
+        auth_spec = spec_manager.get_spec_for_path_and_method(auth_path, auth_method)
+        assert auth_spec is not None, f"Should find spec for {auth_method} {auth_path}"
+        assert "operationId" in auth_spec, "Auth spec should include operationId"
+
+        # 6. Test get_spec_part
+        # Get a specific schema
+        schema = spec_manager.get_spec_part("components", "schemas", "FunctionResponse")
+        assert schema is not None, "Should find FunctionResponse schema"
+        assert "properties" in schema, "Schema should have properties"
+
+        # 7. Test caching behavior
+        # Call get_spec again - should use cached version
+        import time
+
+        start_time = time.time()
+        await spec_manager.get_spec()
+        end_time = time.time()
+        assert (end_time - start_time) < 0.1, "Cached spec retrieval should be fast"