Goro commited on
Commit
3476d1b
·
1 Parent(s): cf52057

Add tests when custom path is provided

Browse files
src/fastmcp/contrib/component_manager/component_manager.py CHANGED
@@ -127,7 +127,10 @@ def make_route(action, component, config, required_scopes, root_path) -> Route:
127
  ]:
128
  path = f"/{{{config['param']}}}/{action}"
129
  else:
130
- path = f"/{component}s/{{{config['param']}}}/{action}"
 
 
 
131
 
132
  return Route(path, endpoint=endpoint, methods=["POST"])
133
 
 
127
  ]:
128
  path = f"/{{{config['param']}}}/{action}"
129
  else:
130
+ if root_path != "/" and required_scopes is None:
131
+ path = f"{root_path}/{component}s/{{{config['param']}}}/{action}"
132
+ else:
133
+ path = f"/{component}s/{{{config['param']}}}/{action}"
134
 
135
  return Route(path, endpoint=endpoint, methods=["POST"])
136
 
tests/contrib/test_component_manager.py CHANGED
@@ -538,3 +538,206 @@ class TestAuthComponentManagementRoutes:
538
  assert response.status_code == 200
539
  assert response.json() == {"message": "Disabled prompt: test_prompt"}
540
  assert prompt.enabled is False
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
538
  assert response.status_code == 200
539
  assert response.json() == {"message": "Disabled prompt: test_prompt"}
540
  assert prompt.enabled is False
541
+
542
+
543
+ class TestComponentManagerWithPath:
544
+ """Test component manager routes when mounted at a custom path."""
545
+
546
+ @pytest.fixture
547
+ def mcp_with_path(self):
548
+ mcp = FastMCP("TestServerWithPath")
549
+ set_up_component_manager(server=mcp, path="/test")
550
+
551
+ @mcp.tool
552
+ def test_tool() -> str:
553
+ return "test_tool_result"
554
+
555
+ @mcp.resource("data://test_resource")
556
+ def test_resource() -> str:
557
+ return "test_resource_result"
558
+
559
+ @mcp.prompt
560
+ def test_prompt() -> str:
561
+ return "test_prompt_result"
562
+
563
+ return mcp
564
+
565
+ @pytest.fixture
566
+ def client_with_path(self, mcp_with_path):
567
+ return TestClient(mcp_with_path.http_app())
568
+
569
+ @pytest.mark.asyncio
570
+ async def test_enable_tool_route_with_path(self, client_with_path, mcp_with_path):
571
+ tool = await mcp_with_path._tool_manager.get_tool("test_tool")
572
+ tool.enabled = False
573
+ response = client_with_path.post("/test/tools/test_tool/enable")
574
+ assert response.status_code == status.HTTP_200_OK
575
+ assert response.json() == {"message": "Enabled tool: test_tool"}
576
+ tool = await mcp_with_path._tool_manager.get_tool("test_tool")
577
+ assert tool.enabled is True
578
+
579
+ @pytest.mark.asyncio
580
+ async def test_disable_resource_route_with_path(
581
+ self, client_with_path, mcp_with_path
582
+ ):
583
+ resource = await mcp_with_path._resource_manager.get_resource(
584
+ "data://test_resource"
585
+ )
586
+ resource.enabled = True
587
+ response = client_with_path.post("/test/resources/data://test_resource/disable")
588
+ assert response.status_code == status.HTTP_200_OK
589
+ assert response.json() == {"message": "Disabled resource: data://test_resource"}
590
+ resource = await mcp_with_path._resource_manager.get_resource(
591
+ "data://test_resource"
592
+ )
593
+ assert resource.enabled is False
594
+
595
+ @pytest.mark.asyncio
596
+ async def test_enable_prompt_route_with_path(self, client_with_path, mcp_with_path):
597
+ prompt = await mcp_with_path._prompt_manager.get_prompt("test_prompt")
598
+ prompt.enabled = False
599
+ response = client_with_path.post("/test/prompts/test_prompt/enable")
600
+ assert response.status_code == status.HTTP_200_OK
601
+ assert response.json() == {"message": "Enabled prompt: test_prompt"}
602
+ prompt = await mcp_with_path._prompt_manager.get_prompt("test_prompt")
603
+ assert prompt.enabled is True
604
+
605
+
606
+ class TestComponentManagerWithPathAuth:
607
+ """Test component manager routes with auth when mounted at a custom path."""
608
+
609
+ def setup_method(self):
610
+ # Generate a key pair and create an auth provider
611
+ key_pair = RSAKeyPair.generate()
612
+ self.auth = BearerAuthProvider(
613
+ public_key=key_pair.public_key,
614
+ issuer="https://dev.example.com",
615
+ audience="my-dev-server",
616
+ required_scopes=["tool:write", "tool:read"],
617
+ )
618
+ self.mcp = FastMCP("TestServerWithPathAuth", auth=self.auth)
619
+ set_up_component_manager(
620
+ server=self.mcp, path="/test", required_scopes=["tool:write", "tool:read"]
621
+ )
622
+ self.token = key_pair.create_token(
623
+ subject="dev-user",
624
+ issuer="https://dev.example.com",
625
+ audience="my-dev-server",
626
+ scopes=["tool:read", "tool:write"],
627
+ )
628
+ self.token_without_scopes = key_pair.create_token(
629
+ subject="dev-user",
630
+ issuer="https://dev.example.com",
631
+ audience="my-dev-server",
632
+ scopes=[],
633
+ )
634
+
635
+ @self.mcp.tool
636
+ def test_tool() -> str:
637
+ return "test_tool_result"
638
+
639
+ @self.mcp.resource("data://test_resource")
640
+ def test_resource() -> str:
641
+ return "test_resource_result"
642
+
643
+ @self.mcp.prompt
644
+ def test_prompt() -> str:
645
+ return "test_prompt_result"
646
+
647
+ self.client = TestClient(self.mcp.http_app())
648
+
649
+ @pytest.mark.asyncio
650
+ async def test_unauthorized_enable_tool(self):
651
+ tool = await self.mcp._tool_manager.get_tool("test_tool")
652
+ tool.enabled = False
653
+ response = self.client.post("/test/tools/test_tool/enable")
654
+ assert response.status_code == 401
655
+ assert tool.enabled is False
656
+
657
+ @pytest.mark.asyncio
658
+ async def test_forbidden_enable_tool(self):
659
+ tool = await self.mcp._tool_manager.get_tool("test_tool")
660
+ tool.enabled = False
661
+ response = self.client.post(
662
+ "/test/tools/test_tool/enable",
663
+ headers={"Authorization": "Bearer " + self.token_without_scopes},
664
+ )
665
+ assert response.status_code == 403
666
+ assert tool.enabled is False
667
+
668
+ @pytest.mark.asyncio
669
+ async def test_authorized_enable_tool(self):
670
+ tool = await self.mcp._tool_manager.get_tool("test_tool")
671
+ tool.enabled = False
672
+ response = self.client.post(
673
+ "/test/tools/test_tool/enable",
674
+ headers={"Authorization": "Bearer " + self.token},
675
+ )
676
+ assert response.status_code == 200
677
+ assert response.json() == {"message": "Enabled tool: test_tool"}
678
+ tool = await self.mcp._tool_manager.get_tool("test_tool")
679
+ assert tool.enabled is True
680
+
681
+ @pytest.mark.asyncio
682
+ async def test_unauthorized_disable_resource(self):
683
+ resource = await self.mcp._resource_manager.get_resource("data://test_resource")
684
+ resource.enabled = True
685
+ response = self.client.post("/test/resources/data://test_resource/disable")
686
+ assert response.status_code == 401
687
+ assert resource.enabled is True
688
+
689
+ @pytest.mark.asyncio
690
+ async def test_forbidden_disable_resource(self):
691
+ resource = await self.mcp._resource_manager.get_resource("data://test_resource")
692
+ resource.enabled = True
693
+ response = self.client.post(
694
+ "/test/resources/data://test_resource/disable",
695
+ headers={"Authorization": "Bearer " + self.token_without_scopes},
696
+ )
697
+ assert response.status_code == 403
698
+ assert resource.enabled is True
699
+
700
+ @pytest.mark.asyncio
701
+ async def test_authorized_disable_resource(self):
702
+ resource = await self.mcp._resource_manager.get_resource("data://test_resource")
703
+ resource.enabled = True
704
+ response = self.client.post(
705
+ "/test/resources/data://test_resource/disable",
706
+ headers={"Authorization": "Bearer " + self.token},
707
+ )
708
+ assert response.status_code == 200
709
+ assert response.json() == {"message": "Disabled resource: data://test_resource"}
710
+ resource = await self.mcp._resource_manager.get_resource("data://test_resource")
711
+ assert resource.enabled is False
712
+
713
+ @pytest.mark.asyncio
714
+ async def test_unauthorized_enable_prompt(self):
715
+ prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
716
+ prompt.enabled = False
717
+ response = self.client.post("/test/prompts/test_prompt/enable")
718
+ assert response.status_code == 401
719
+ assert prompt.enabled is False
720
+
721
+ @pytest.mark.asyncio
722
+ async def test_forbidden_enable_prompt(self):
723
+ prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
724
+ prompt.enabled = False
725
+ response = self.client.post(
726
+ "/test/prompts/test_prompt/enable",
727
+ headers={"Authorization": "Bearer " + self.token_without_scopes},
728
+ )
729
+ assert response.status_code == 403
730
+ assert prompt.enabled is False
731
+
732
+ @pytest.mark.asyncio
733
+ async def test_authorized_enable_prompt(self):
734
+ prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
735
+ prompt.enabled = False
736
+ response = self.client.post(
737
+ "/test/prompts/test_prompt/enable",
738
+ headers={"Authorization": "Bearer " + self.token},
739
+ )
740
+ assert response.status_code == 200
741
+ assert response.json() == {"message": "Enabled prompt: test_prompt"}
742
+ prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
743
+ assert prompt.enabled is True