| import os |
| import sys |
| import types |
| import unittest |
|
|
|
|
| os.environ.setdefault("MARINE_API_URL", "http://example.invalid") |
|
|
| if "httpx" not in sys.modules: |
| httpx = types.ModuleType("httpx") |
| httpx.Client = object |
| httpx.Timeout = object |
| sys.modules["httpx"] = httpx |
|
|
| if "mcp.server.mcpserver" not in sys.modules: |
| mcp_package = types.ModuleType("mcp") |
| mcp_server_package = types.ModuleType("mcp.server") |
| mcp_server_module = types.ModuleType("mcp.server.mcpserver") |
|
|
| class MCPServer: |
| def __init__(self, *args, **kwargs): |
| pass |
|
|
| def tool(self): |
| return lambda function: function |
|
|
| def run(self): |
| pass |
|
|
| mcp_server_module.MCPServer = MCPServer |
| sys.modules["mcp"] = mcp_package |
| sys.modules["mcp.server"] = mcp_server_package |
| sys.modules["mcp.server.mcpserver"] = mcp_server_module |
|
|
| import marine_mcp |
|
|
|
|
| class FisheriesInventoryAliasTests(unittest.TestCase): |
| def setUp(self): |
| self.original_hf_files = marine_mcp._hf_files |
|
|
| def fake_hf_files(domain="all", force=False): |
| rows = [ |
| { |
| "type": "file", |
| "path": "IATTC/sample.csv", |
| "size": 120, |
| "repository": marine_mcp.HF_TUNA_DATASET_REPO, |
| "repository_domain": "tuna", |
| }, |
| { |
| "type": "file", |
| "path": "current/sprfmo_effort.csv", |
| "size": 80, |
| "repository": marine_mcp.HF_SQUID_DATASET_REPO, |
| "repository_domain": "squid", |
| }, |
| ] |
| if domain in {"tuna", "squid"}: |
| rows = [x for x in rows if x["repository_domain"] == domain] |
| return rows, {} |
|
|
| marine_mcp._hf_files = fake_hf_files |
|
|
| def tearDown(self): |
| marine_mcp._hf_files = self.original_hf_files |
|
|
| def test_query_alias_becomes_keyword(self): |
| result = marine_mcp.fisheries_inventory(query="IATTC") |
| self.assertEqual(result["matched_file_count"], 1) |
| self.assertEqual(result["keyword"], "IATTC") |
|
|
| def test_source_alias_becomes_keyword(self): |
| result = marine_mcp.fisheries_inventory(source="IATTC") |
| self.assertEqual(result["matched_file_count"], 1) |
| self.assertEqual(result["keyword"], "IATTC") |
|
|
| def test_source_tuna_becomes_domain(self): |
| result = marine_mcp.fisheries_inventory(source="tuna") |
| self.assertEqual(result["domain"], "tuna") |
| self.assertEqual(result["matched_file_count"], 1) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
