File size: 2,654 Bytes
95a8a23 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | 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()
|